feat(scheduler-core): implement record start/stop lifecycle (task 9.2)

Adds recording-state fields directly to xce_scheduler_t (no new xce_sched_driver_t type, per task 9.1's precedent). sched_record_start()/ sched_record_stop() implement AC 10: validate via task 9.1's path validator, open with a TOCTOU-hardened sequence (O_NOFOLLOW, no O_TRUNC, fstat()+st_nlink==1 check before ftruncate()), snapshot the core, write ReplayHeader+SnapshotEntry on start, ReplayEvent{endSession} on stop. driver.c gains boolean-gated recording hooks in xce_sched_driver_translate_pending_events() (per-event-batch ReplayEvent writes, new recording_events.c) and xce_sched_driver_run_normal_mode_round() (per-session ReplaySession writes, new recording_session.c) -- zero I/O/ allocation when inactive. Two real bugs found and fixed during implementation: bug-1498 (O_TRUNC would have truncated a hard-linked victim file before the TOCTOU check could reject it) and bug-1499 (a transient edit dropped xce_core_signal_go(), causing a 3-way thread deadlock). test_record_lifecycle.c decodes the full start->events->session->stop cycle with the real capnp decoders and asserts the exact message sequence; TOCTOU regression tests use real symlink(2)/link(2) syscalls. 25/25 scheduler suite, clang-format clean.
2026-08-05 09:07:51 +00:00 · 2026-08-05 09:07:51 +00:00 · ff8d4f416e
commit ff8d4f416e
parent 4e5547ee55
15 changed files with 2110 additions and 36 deletions
--- a/.kiro/specs/scheduler-core/tasks.md
+++ b/.kiro/specs/scheduler-core/tasks.md
@ -224,7 +224,7 @@
  - _Depends: 8.1_
  - _Verified: 2026-08-04 — `--record-dir` CLI flag added to main.c/scheduler.h (record_dir field, never a request-path field). sched_validate_record_filename() (new recording.c/recording.h) rejects any `filename` containing `/` or `..` before any filesystem call, then resolves via realpath() and confirms canonical (not textual-prefix) containment beneath record_dir via path_is_contained(); rejects pre-existing non-regular entries (-EEXIST) and a non-directory record_dir (-ENOTDIR). Adversarial security review found and this task fixed two real CWE-59 inode-aliasing bypasses across 3 rounds: round 1 found a direct hard link inside record_dir aliasing an outside file passed validation (S_ISREG doesn't distinguish a hard-linked regular file from a genuine one) — fixed with an entry_st.st_nlink > 1 check (-EACCES) in the direct-entry branch, live RED→GREEN reproduced. Round 2 re-review found a compound variant still open: a symlink inside record_dir pointing at an in-directory file itself hard-linked outside record_dir was still accepted, since the S_ISLNK branch checked containment but never st_nlink on the resolved target — fixed by stat()'ing the already-realpath()-resolved target and applying the same S_ISREG && st_nlink>1 check, generalizing correctly to multi-hop symlink chains, live RED→GREEN reproduced. Round 3 re-review independently reproduced both fixes via a standalone PoC harness and additionally hunted for further variants (2-hop chains, symlink-to-symlink escape, FIFO/socket direct-naming) — all clean, no new bypass found; one non-blocking observation logged (symlink resolving to a contained-but-non-regular target is accepted rather than rejected, pre-existing since round 1, not the CWE-22/CWE-59 escape class, already covered by task 9.2's planned O_NOFOLLOW hardening) — APPROVED. TOCTOU residual risk (validation-to-eventual-fopen() window) explicitly documented in recording.h and deferred to task 9.2 (O_NOFOLLOW + fstat()-based st_nlink==1 re-check on the opened fd), not this task's scope. 18/18 recording subtests (16 original + 2 hard-link/compound regressions), 24/24 scheduler suite, clang-format clean — all independently re-verified by me via a fresh `meson compile`/`meson test --suite scheduler` run on current HEAD before marking complete._
- [ ] 9.2 Implement the record start/stop lifecycle
+- [x] 9.2 Implement the record start/stop lifecycle
  - On start: take a full snapshot, write the replay header followed by the initial snapshot entry, begin tracking recording state
  - On each event batch and each completed session while recording: write the corresponding event/session messages to the record file in order — extends the normal-mode driver main loop built in group 5 with recording insert points, it does not replace it
  - On stop: write the end-of-recording marker, flush and close the file, clear recording state
@ -232,6 +232,7 @@
  - Observable: a start → submit events → complete a session → stop cycle produces a file beginning with header+snapshot-entry and ending with the end-of-recording marker, with events and one session in between, byte-decodable by the round-trip decoder from group 7
  - _Requirements: 5.10_
  - _Depends: 9.1_
  - _Verified: 2026-08-05 — added recording-state fields directly to xce_scheduler_t (no new xce_sched_driver_t type, matching task 9.1's established deviation from design.md's sketch). sched_record_start()/sched_record_stop() (recording.c) implement AC 10 steps 1 and 7 exactly: validate via sched_validate_record_filename() (task 9.1), open with the TOCTOU-hardened sequence (O_NOFOLLOW, no O_TRUNC, fstat()+st_nlink==1 check on the opened fd, ftruncate() only after the check passes — matching recording.h's task-9.1-era note to this task), xce_sched_core_save_snapshot() → write ReplayHeader then SnapshotEntry, set recording=true; stop writes ReplayEvent{body=endSession}, flushes, closes, clears state. driver.c gains a record_ctx void* param on xce_sched_driver_translate_pending_events() (deliberately NOT read from core->driver_ctx, which isn't guaranteed xce_scheduler_t* in test fixtures) for per-event-batch ReplayEvent writes (new recording_events.c), plus sched_record_session_begin()/_add_decision()/_end() hooks in xce_sched_driver_run_normal_mode_round() for per-session ReplaySession writes (new recording_session.c) — both boolean-gated on sched->recording, zero I/O/allocation when inactive, verified by tracing both call sites. recording.c split three ways (recording.c=543, recording_events.c=177, recording_session.c=244 lines) to stay under the 800-line cap, mirroring snapshot.c's precedent. Two real bugs were found and fixed during implementation, independently confirmed by review: bug-1498 (an initial O_TRUNC-in-open() would have truncated a hard-linked victim file before the st_nlink check could reject it — fixed by deferring truncation to a post-check ftruncate()) and bug-1499 (an edit transiently dropped xce_core_signal_go(), a 3-way picker/allocator/driver deadlock — restored, caught by test_driver_session_ordering.c). New test_record_lifecycle.c (826 lines, in line with this suite's established over-800 precedent for test files) decodes the full start→events→session→stop cycle with the real generated capnp decoders and asserts the exact 5-message sequence plus no 6th message; two TOCTOU regression tests use real symlink(2)/link(2) syscalls against a real victim file and assert its content is untouched after rejection; double-start/stop-when-not-recording both return -EALREADY cleanly. Review found one non-blocking follow-up (open() lacks O_NONBLOCK, so a FIFO planted in the same TOCTOU window would hang the calling thread — DoS, not corruption; logged below, not fixed in this task) and confirmed zero scope leakage into tasks 9.3/9.4/group 10. 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. Note: two implementer-agent runs ended up editing this worktree concurrently after a spurious "agent died" signal turned out to be wrong; a stabilization wait (repeated md5sum) plus two independent fresh-build verifications (mine and a dedicated fix/verify agent's) confirmed the settled state was consistent before review._
 - [ ] 9.3 Implement periodic snapshot checkpointing and recording error handling
  - Append a fresh snapshot entry to the record file every configured number of recorded events (0 disables this)
@ -382,3 +383,5 @@
 - Task 4.3 split its implementation into a NEW file `core_events.c` (not `core.c`, which design.md's File Structure Plan literally names) because implementing `xce_core_push_events()` directly in `core.c` would have pushed it to 977 lines, over this project's 800-line hard cap. `core.h` (the public contract) matches design.md's intent; only the `.c` file location differs. Worth reflecting in design.md next time it's touched. Also: `test_core_isolation.c`'s nm(1)-based zero-controller-I/O boundary audit (Req 5 AC 3) covers only `core.c`, not `core_events.c` — manually verified clean (no netio/xceapi/client/delay/idgen usage) but not mechanically enforced; extending that test to cover `core_events.c` would be a reasonable low-risk follow-up before later tasks add more logic to this file.
 - Same pattern compounded across tasks 4.4 (`core_session.c`) and 4.5 (`core_clock.c`, which also genuinely NEEDS to be outside the wall-clock-call scan since it's the one file allowed to call `clock_gettime`) — `test_core_isolation.c` now covers only `core.c`, while three sibling files (`core_events.c`/`core_session.c`/`core_clock.c`) sit outside its mechanical audit, each manually verified clean but not enforced by a test. Worth a dedicated follow-up task extending the forbidden-header/forbidden-symbol half of the audit (not the wall-clock half, which `core_clock.c` must stay exempt from) to all `core_*.c` files before Phase 3 adds more.
 - Task 5.2's review surfaced a genuine, pre-existing bug in `plugin.c`'s `plugin_end_session()` (predates this entire spec — `git blame` traces it to commit `638173dec`, 2026-02-26; untouched by any scheduler-core task's diff): the per-plugin guard checks `plugin->begin_session_fn != NULL` but then calls `plugin->end_session_fn(plugin)` — a guard/call mismatch. Confirmed live impact: `plugins/fcfs.c` registers `.begin_session_fn = NULL`/`.end_session_fn = fcfs_end_session`, so `fcfs_end_session()` (resets the legacy `job_elect_fn`-style iteration cursor) is never invoked today. Only matters if a deployment's `job_pick_plugin` config selects the legacy `xce_sched_fcfs` name rather than the new default `builtin_fcfs` (both fields NULL on `builtin_fcfs`, unaffected). Out of scope to fix here — needs its own bug-log entry / small fix task, not folded into scheduler-core.
 - Task 9.2's review found a non-blocking follow-up, not fixed in that task: `sched_record_start()`'s TOCTOU-hardened `open()` (`recording.c`, `O_WRONLY|O_CREAT|O_NOFOLLOW`) lacks `O_NONBLOCK`. A FIFO planted at the resolved path in the same race window the symlink/hard-link defenses already close (between `sched_validate_record_filename()` returning and `open()` running) would hang the calling thread indefinitely — a DoS (thread hang), not data corruption, since the existing `st_nlink`/`S_ISREG` checks still reject it once `open()` eventually returns. Adding `O_NONBLOCK` to the open flags is zero-risk for the regular-file case and closes this; worth a small follow-up before task 9.2's code sees production traffic.
 - Cross-task interaction, flagged before task 9.4 exists — resolution decided (not a fail/retry, a deferral): task 9.2's `sched_record_start()` calls `xce_sched_core_save_snapshot()` as its first action (AC 10 step 1) — the same function task 8.2 gated to return `-EBUSY` whenever `session_active != 0`. `main.c`'s driver loop is single-threaded (`while(1) { xce_sched_driver_run_normal_mode_round(&sched); ...; sleep(1); }`); incoming requests (including the future `SchedAdminRequest`) arrive asynchronously via `scheduler_req_handler` (handler.c) on the I/O side, same as job/queue/worker updates. **Decision**: task 9.4 must NOT call `sched_record_start()` directly from the request-handling thread (that both risks racing `session_active` and violates this codebase's async-I/O rule against blocking/file-I/O in a request-handler callback). Instead, `recordStart` must be enqueued the same way job/queue/worker updates already are (`pend_updates` dataq or an equivalent dedicated pending-admin-request slot) and picked up by `xce_sched_driver_translate_pending_events()` — which already runs at the top of every round, strictly before `begin_session()`, so it is guaranteed between-sessions by the same construction argument design.md uses for `core_push_events()`. This means recording "takes effect" at the next scheduling round, not synchronously on request arrival. Consequence for task 9.4: since AC 10 specifies a synchronous request-reply carrying the resolved filename, the actual `SchedAdminReply` send must be deferred until the driver thread performs the queued start (not sent eagerly from the handler thread) — needs a completion signal back to the waiting request context (e.g. a condition variable, or replying once `xce_sched_driver_translate_pending_events()` reports the deferred start applied). Given this deferral makes the direct `-EBUSY` race effectively unreachable from the admin path, the AC 10 error-codes table doesn't need a new "busy" entry for this case — `sched_record_start()` keeps its `-EBUSY` return as a defensive/fail-loud contract check (task 9.2's job, no torn state), it just should no longer be a path task 9.4's dispatch can actually trigger by construction.
--- a/src/scheduler/include/driver.h
+++ b/src/scheduler/include/driver.h
@ -49,8 +49,26 @@
 * translation batch (rare, defensive path); or `xce_core_push_events()`'s
 * own negative return (e.g. -EBUSY if a session was unexpectedly active)
 * passed straight through.
 *
 * `record_ctx` (scheduler-core spec, task 9.2, Req 5 AC 10): an opaque
 * `void*`, cast back to `xce_scheduler_t*` inside driver.c, for the per-
 * event-batch recording-write hook -- same `void*`-instead-of-
 * `xce_scheduler_t*` idiom xce_sched_driver_on_decision()/
 * xce_sched_driver_run_normal_mode_round() below already use, and for the
 * same reason (this file's boundary comment: depends on core.h only).
 * Pass NULL to disable recording for this call (zero I/O, zero extra
 * allocation, Req 5 AC 10 "Performance") -- every existing caller that
 * does not care about recording does exactly this.
 *
 * Deliberately a SEPARATE parameter from `core->driver_ctx`
 * (xce_core_set_decision_cb()'s decision-callback context): that field is
 * only an `xce_scheduler_t*` in production wiring (main.c) -- some tests
 * register it as a different, smaller struct for their own decision-
 * callback spy, so this function must not assume it is safe to
 * reinterpret as `xce_scheduler_t*`.
 */
-extern int xce_sched_driver_translate_pending_events(xce_sched_core_t* core);
+extern int xce_sched_driver_translate_pending_events(xce_sched_core_t* core,
                                                     void* record_ctx);
 /*
 * xce_sched_driver_on_decision - the normal-mode driver's decision-publish
--- a/src/scheduler/include/recording.h
+++ b/src/scheduler/include/recording.h
@ -3,16 +3,21 @@
 #define _RECORDING_H_ 1
 #include <stddef.h>
 #include <stdint.h>
 #include "scheduler.h"
 /*
- * recording.h - scheduler-core spec, task 9.1 (Req 5 AC 10, "Record file
+ * recording.h - scheduler-core spec, tasks 9.1/9.2 (Req 5 AC 10, "Record
- * path validation"). First piece of design.md's File Structure Plan
+ * file path validation" and "Runtime recording control -- detailed
- * `recording.c` ("NEW (Phase 3): record start/stop/status, ReplayEvent/
+ * mechanism"). Task 9.1 added ONLY sched_validate_record_filename() below,
- * ReplaySession writes") -- this task adds ONLY the validation function
+ * the validation function every write must pass through first. Task 9.2
- * every future write must pass through first; the actual record start/
+ * (this revision) adds the actual record start/stop lifecycle
- * stop/status lifecycle (design.md's NormalDriver service interface --
+ * (sched_record_start()/sched_record_stop()) plus the recording-write hooks
- * sched_driver_record_start()/_stop()/_status(), `xce_sched_driver_t`'s
+ * driver.c's normal-mode loop calls into -- design.md's NormalDriver
- * recording-state fields) is task 9.2's job and is NOT implemented here.
+ * service interface, adapted to this codebase's single combined
 * `xce_scheduler_t` driver-level struct (there is no separate
 * `xce_sched_driver_t` -- see scheduler.h's field comment on `recording`).
 *
 * sched_validate_record_filename - validates a caller-supplied, admin-
 * authenticated but UNTRUSTED record file name (controller.capnp's
@ -113,19 +118,155 @@
 * rejection in step 1 returns -EINVAL without a single filesystem call,
 * even if `record_dir` itself does not exist.
 *
- * NOTE for task 9.2 (not implemented here -- this task is validation-only):
+ * TOCTOU note (closed by sched_record_start() below, task 9.2): this
- * this function's containment/aliasing checks are only valid up to the
+ * function's containment/aliasing checks are only valid up to the moment it
- * moment it returns -- a TOCTOU race could still swap a symlink or hard
+ * returns -- a TOCTOU race could still swap a symlink or hard link into
- * link into `out_path` between this call returning and task 9.2's eventual
+ * `out_path` between this call returning and the eventual fopen()/open().
- * fopen(). When task 9.2 opens `out_path` for writing, it should (a) pass
+ * sched_record_start() closes that window by (a) passing O_NOFOLLOW so a
- * O_NOFOLLOW so a symlink swapped in after validation is refused at
+ * symlink swapped in after validation is refused at open() rather than
- * open() rather than followed, and (b) fstat() the resulting fd and
+ * followed, and (b) fstat()-ing the resulting fd and rejecting unless
- * reject unless st_nlink == 1, since O_NOFOLLOW alone does not defend
+ * st_nlink == 1, since O_NOFOLLOW alone does not defend against a hard link
- * against a hard link swapped in after validation (no symlink is involved
+ * swapped in after validation (no symlink is involved in that race, so
- * in that race, so O_NOFOLLOW never triggers).
+ * O_NOFOLLOW never triggers).
 */
 extern int sched_validate_record_filename(const char* record_dir,
                                          const char* filename, char* out_path,
                                          size_t out_path_len);
 /*
 * sched_record_start - Req 5 AC 10 (scheduler-core spec, task 9.2,
 * "Recording lifecycle -- On record start"). Begins a new recording:
 *
 *   1. Resolves/validates `record_file` (a bare filename, or NULL/"" to
 *      auto-generate one) via sched_validate_record_filename() above,
 *      against `sched->record_dir`.
 *   2. Opens the resolved path with the TOCTOU-hardened sequence this
 *      header's own doc comment describes (O_NOFOLLOW + fstat()+
 *      st_nlink==1 check) -- closes the race window
 *      sched_validate_record_filename() itself flags as still-open after
 *      its own return.
 *   3. Takes a full core snapshot (xce_sched_core_save_snapshot(),
 *      core.h) and writes `ReplayHeader` followed by the initial
 *      `SnapshotEntry` as the first two messages of the file (Req 5 AC 20:
 *      "the first data message after ReplayHeader MUST be a
 *      SnapshotEntry").
 *   4. On success, sets `sched->recording = true`, stores the resolved
 *      path in `sched->record_file_path` (strdup'd), resets
 *      `sched->record_event_count = 0`, and stamps
 *      `sched->record_start_ts = xce_core_now(&sched->core)`.
 *
 * Preconditions: fails with -EALREADY if `sched->recording` is already
 * true (design.md's NormalDriver contract) -- no file is touched. Fails
 * with -EINVAL if `sched` is NULL.
 *
 * Postconditions: on success, `ReplayHeader` + the initial `SnapshotEntry`
 * are durably on disk (fwrite completed) before this function returns.
 * On any failure after opening the file (write error, snapshot failure),
 * the partially-written file is closed and this function returns a
 * negative errno WITHOUT setting `sched->recording` -- the caller may
 * retry with a fresh sched_record_start() call. Never leaves `recording`
 * true with a NULL `record_file`, or vice versa (scheduler.h's own
 * invariant comment on `xce_scheduler_t.recording`).
 *
 * `record_file` may be NULL or empty to auto-generate a timestamped name
 * (`scheduler-replay-<UTC timestamp>.replay`) beneath `sched->record_dir`.
 */
 extern int sched_record_start(xce_scheduler_t* sched, const char* record_file);
 /*
 * sched_record_stop - Req 5 AC 10 (scheduler-core spec, task 9.2,
 * "Recording lifecycle -- On record stop"). Ends the current recording:
 * writes a `ReplayEvent{body=endSession}` message (the end-of-recording
 * marker -- there is no separate top-level footer message type, see
 * requirements.md Req 5 AC 20), then fflush()+fclose()s `sched->record_file`
 * and clears `sched->recording`/`sched->record_file`/`sched->record_file_path`
 * (freeing the latter) regardless of whether the marker write succeeded
 * (matches Req 5 AC 10's "Error handling" clause: a failed record write
 * disables recording rather than leaving it in a half-broken active state).
 *
 * Preconditions: fails with -EALREADY, no state touched, if
 * `sched->recording` is already false. Fails with -EINVAL if `sched` is
 * NULL.
 *
 * Returns 0 on a clean stop; a negative errno if the end-of-recording
 * 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).
 */
 extern int sched_record_stop(xce_scheduler_t* sched);
 /*
 * sched_record_write_events - the per-event-batch recording-write hook
 * (Req 5 AC 10, "On each core_push_events() call ... For each event in the
 * batch, serialize and write the corresponding ReplayEvent message").
 * Called from xce_sched_driver_translate_pending_events() (driver.c) once
 * per drained batch, immediately after xce_core_push_events() and BEFORE
 * that function's own cleanup loop frees any translated payload (so every
 * pointer inside `events` is still guaranteed live).
 *
 * Boolean-gated: a NULL `sched`, a NULL/empty `events`, or
 * `sched->recording == false` is a zero-cost no-op (Req 5 AC 10,
 * "Performance"). On a write failure mid-batch, recording is disabled
 * (mirrors sched_record_stop()'s cleanup, minus the end-of-recording
 * marker -- the file is presumably broken already) and a negative errno is
 * returned; the caller (driver.c) does not need to react further -- normal
 * scheduling continues either way (Req 5 AC 10, "Error handling").
 */
 extern int sched_record_write_events(xce_scheduler_t*  sched,
                                     xce_core_event_t* events, int n);
 /*
 * sched_record_session_ctx_t - opaque per-round accumulator for the
 * decisions a recording session collects between
 * sched_record_session_begin()/sched_record_session_end() (Req 5 AC 10,
 * "On core_go(): begin building a ReplaySession message" /
 * "In the decision drain loop: append a ReplayDecision" /
 * "On core_end_session(): finalize ... write the complete message").
 * Defined privately in recording.c; driver.c only ever holds a pointer.
 */
 typedef struct sched_record_session_ctx_s sched_record_session_ctx_t;
 /*
 * sched_record_session_begin - starts accumulating a ReplaySession for one
 * normal-mode round. Returns NULL (zero allocation) when `sched` is NULL,
 * `sched->recording` is false, or `sched->record_file` is NULL -- the
 * caller (xce_sched_driver_run_normal_mode_round(), driver.c) then simply
 * never calls sched_record_session_add_decision()/sched_record_session_end(),
 * satisfying Req 5 AC 10's "zero extra allocation when recording == false".
 * On success, returns a heap-allocated context stamped with
 * `begin_round_ts` (the ReplaySession.beginRound field).
 */
 extern sched_record_session_ctx_t*
 sched_record_session_begin(xce_scheduler_t* sched, uint64_t begin_round_ts);
 /*
 * sched_record_session_add_decision - appends one drained decision to
 * `ctx`'s in-progress ReplaySession.decisions list, tagged with `ts`. A
 * NULL `ctx` (recording inactive -- see sched_record_session_begin() above)
 * or NULL `decision` is a no-op. XCE_DECISION_PREEMPT is not yet supported
 * (mirrors driver.c's xce_sched_driver_on_decision() -- no live allocator
 * path produces one, and this is therefore currently unreachable, but is
 * handled loudly (ERROR log) rather than silently dropped or crashing) --
 * see driver.c for the shared rationale. XCE_DECISION_NONE/
 * XCE_DECISION_NEW_SESSION are likewise skipped (nothing to record).
 */
 extern void sched_record_session_add_decision(sched_record_session_ctx_t* ctx,
                                              uint64_t                    ts,
                                              xce_decision_t* decision);
 /*
 * sched_record_session_end - finalizes and writes the accumulated
 * ReplaySession message (begin_round + decisions + a SessionSummary
 * computed from `end_ts` and the tallies sched_record_session_add_decision()
 * collected) to `sched->record_file`, then frees `ctx` and everything it
 * owns. A NULL `ctx` is a no-op (0 is returned; nothing to free). On a
 * write failure, recording is disabled the same way
 * sched_record_write_events() does. Always frees `ctx` before returning,
 * even on failure or when `sched`'s recording state changed out from under
 * the caller between sched_record_session_begin() and this call.
 */
 extern int sched_record_session_end(xce_scheduler_t*            sched,
                                    sched_record_session_ctx_t* ctx,
                                    uint64_t                    end_ts);
 #endif
--- a/src/scheduler/include/scheduler.h
+++ b/src/scheduler/include/scheduler.h
@ -5,7 +5,9 @@
 #include <pthread.h>
 #include <rtti.h>
 #include <stdatomic.h>
 #include <stdbool.h>
 #include <stdint.h>
 #include <stdio.h>
 #include <time.h>
 #include "client.h"
@ -117,6 +119,37 @@ typedef struct {
     * xce_sched_driver_t (record_file_path/record_file/recording/
     * record_event_count/...) is task 9.2's job, not this field's. */
    char* record_dir;
    /* Req 5 AC 10 (scheduler-core spec, task 9.2): the recording lifecycle
     * state design.md sketches on a future xce_sched_driver_t -- per
     * record_dir's own comment above, this codebase keeps a single combined
     * driver-level struct (xce_scheduler_t), so these fields land here
     * instead of a new type. Owned exclusively by recording.c's
     * sched_record_start()/sched_record_stop() (recording.h) -- see those
     * functions' doc comments for the full lifecycle contract (path
     * validation + TOCTOU-hardened open on start, end-of-recording marker +
     * flush/close on stop). driver.c's normal-mode loop (driver.h) reads
     * `recording` to boolean-gate its per-event-batch/per-session recording
     * 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. */
    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
     * scope) answer with the current path even after record_file closes. */
    char* record_file_path;
    /* replay file handle; NULL when not recording. */
    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). */
    uint64_t record_event_count;
    /* N events between periodic snapshot checkpoints (0 = disabled).
     * Reserved for task 9.3 -- inert (never read) as of task 9.2. */
    uint64_t record_snapshot_interval;
    /* logical time (xce_core_now()) recording started; 0 = not recording. */
    uint64_t record_start_ts;
 } xce_scheduler_t;
 extern int  sched_init(xce_scheduler_t*, xce_client_config_t*,
--- a/src/scheduler/meson.build
+++ b/src/scheduler/meson.build
@ -12,6 +12,8 @@ scheduler_srcs = [
  'src/snapshot_load.c',
  'src/decision.c',
  'src/recording.c',
  'src/recording_events.c',
  'src/recording_session.c',
  'src/schedobj.c',
  'src/plugin.c',
  'src/resource.c',
--- a/src/scheduler/src/driver.c
+++ b/src/scheduler/src/driver.c
@ -5,6 +5,7 @@
 #include "controller.capnp.h"
 #include "core.h"
 #include "driver.h"
 #include "recording.h"
 #include "scheduler.h"
 #include "utils.h"
@ -280,7 +281,8 @@ static int translate_one_pending_update(xce_sched_core_t*           core,
    }
 }
-int xce_sched_driver_translate_pending_events(xce_sched_core_t* core)
+int xce_sched_driver_translate_pending_events(xce_sched_core_t* core,
                                              void*             record_ctx)
 {
    dataq_t*                     q;
    data_item_t*                 it;
@ -374,6 +376,34 @@ int xce_sched_driver_translate_pending_events(xce_sched_core_t* core)
    push_rc = xce_core_push_events(core, events, n);
    /* Req 5 AC 10 (scheduler-core spec, task 9.2): per-event-batch
     * recording write. `record_ctx` is an opaque `void*` -- always the
     * owning `xce_scheduler_t*` when the caller wants recording, NULL
     * otherwise -- for the SAME reason xce_sched_driver_on_decision()/
     * xce_sched_driver_run_normal_mode_round() already take an opaque
     * `void*` rather than `xce_scheduler_t*` (this file's boundary comment:
     * driver.h depends on core.h only). Deliberately NOT read from
     * `core->driver_ctx`: that field is the decision-callback context
     * (xce_core_set_decision_cb()'s own `driver_ctx` argument), which is
     * only an `xce_scheduler_t*` in PRODUCTION wiring (main.c) -- some
     * white-box tests in this directory register it as a different,
     * smaller struct for their own decision-callback spy, so reinterpreting
     * it here would be a type-confusion bug, not just an unlikely-NULL
     * case. xce_sched_driver_run_normal_mode_round() below passes its own
     * `sched` parameter (always genuinely `xce_scheduler_t*`) explicitly.
     * A NULL `record_ctx` is the single boolean check + zero I/O + zero
     * allocation non-recording path (Req 5 AC 10, "Performance"). MUST run
     * before the cleanup loop below frees any translated payload --
     * sched_record_write_events() needs every `events[i]` pointer still
     * live. */
    {
        xce_scheduler_t* rec_sched = (xce_scheduler_t*)record_ctx;
        if (rec_sched != NULL) {
            sched_record_write_events(rec_sched, events, n);
        }
    }
    for (i = 0; i < n; i++) {
        release_translated_payload(core, reqs[i], (push_rc >= 0));
        free(reqs[i]);
@ -505,8 +535,9 @@ int xce_sched_driver_on_decision(void* driver_ctx, xce_decision_t* decision)
 */
 void xce_sched_driver_run_normal_mode_round(void* driver_ctx)
 {
-    xce_scheduler_t* sched = (xce_scheduler_t*)driver_ctx;
+    xce_scheduler_t*            sched   = (xce_scheduler_t*)driver_ctx;
-    int              rc;
+    sched_record_session_ctx_t* rec_ctx = NULL;
    int                         rc;
    if (sched == NULL) {
        return;
@ -516,8 +547,11 @@ void xce_sched_driver_run_normal_mode_round(void* driver_ctx)
     * accumulated since the last round. Replaces sched_update()'s inline
     * dataq-swap handling. Only valid between sessions (driver.h's own doc
     * comment) -- called here, before xce_core_begin_session(), exactly
-     * where sched_update() ran relative to sched_begin_session() before. */
+     * where sched_update() ran relative to sched_begin_session() before.
-    xce_sched_driver_translate_pending_events(&(sched->core));
+     * `sched` is passed as the recording context (task 9.2) -- see that
     * function's own comment for why this is NOT read from
     * core->driver_ctx. */
    xce_sched_driver_translate_pending_events(&(sched->core), sched);
    rc = xce_core_begin_session(&(sched->core));
    if (rc < 0) {
@ -531,6 +565,20 @@ void xce_sched_driver_run_normal_mode_round(void* driver_ctx)
    plugin_begin_session(sched);
    sched_account_jobs(sched);
    /* Req 5 AC 10 (scheduler-core spec, task 9.2): "On core_go(): begin
     * building a ReplaySession message with beginRound = xce_core_now(core)"
     * -- requirements.md's own annotated pseudocode places this build step
     * immediately BEFORE sched_go() (xce_core_signal_go() below), not
     * after, so it is placed here. sched_record_session_begin() itself is
     * the single boolean check + zero-allocation gate (Req 5 AC 10,
     * "Performance") -- it returns NULL immediately when `sched->recording`
     * is false, and every hook below is a cheap `rec_ctx != NULL` pointer
     * check with no further cost on that path. */
    if (sched->recording) {
        rec_ctx
            = sched_record_session_begin(sched, xce_core_now(&(sched->core)));
    }
    xce_core_signal_go(&(sched->core));
    for (;;) {
@ -545,6 +593,17 @@ void xce_sched_driver_run_normal_mode_round(void* driver_ctx)
            break;
        }
        /* Req 5 AC 10: "In the decision drain loop ... append a
         * ReplayDecision to the current ReplaySession.decisions list" --
         * captured BEFORE the publish callback below, since the decision's
         * payload pointers must be read while still guaranteed live (see
         * recording_session.c's sched_record_session_add_decision() for the
         * full ownership argument). */
        if (rec_ctx != NULL) {
            sched_record_session_add_decision(
                rec_ctx, xce_core_now(&(sched->core)), decision);
        }
        if (sched->core.on_decision != NULL) {
            rc = sched->core.on_decision(sched->core.driver_ctx, decision);
            TRACE("%s rc %d", __func__, rc);
@ -557,4 +616,10 @@ void xce_sched_driver_run_normal_mode_round(void* driver_ctx)
    plugin_end_session(sched);
    xce_core_end_session(&(sched->core));
    /* Req 5 AC 10: "On core_end_session(): finalize the ReplaySession with
     * session summary, write the complete message to the .replay file". */
    if (rec_ctx != NULL) {
        sched_record_session_end(sched, rec_ctx, xce_core_now(&(sched->core)));
    }
 }
--- a/src/scheduler/src/recording.c
+++ b/src/scheduler/src/recording.c
@ -1,18 +1,33 @@
 #include <errno.h>
 #include <fcntl.h>
 #include <limits.h>
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
 #include <sys/stat.h>
 #include <time.h>
 #include <unistd.h>
 #include "controller.capnp.h"
 #include "core.h"
 #include "recording.h"
 #include "utils.h"
 /*
- * recording.c - scheduler-core spec, task 9.1. See recording.h for the
+ * recording.c - scheduler-core spec, tasks 9.1/9.2. See recording.h for the
- * full contract/rationale. This is deliberately the ONLY function in this
+ * full per-function contract/rationale. Task 9.1 added ONLY
- * file as of task 9.1 -- record start/stop/status (task 9.2, design.md's
+ * sched_validate_record_filename() below. Task 9.2 (this revision) adds
- * NormalDriver) extend this file later; nothing here anticipates that
+ * the record start/stop lifecycle (sched_record_start()/sched_record_stop())
- * work's shape beyond what this task's own scope requires.
+ * plus the shared message-write/error-abort helpers
 * (write_capnp_message()/sched_record_abort_on_error()) two sibling files
 * -- recording_events.c (the per-event-batch recording-write hook) and
 * recording_session.c (the per-session ReplaySession accumulator) -- also
 * use. Split across three files (this project's 800-line-per-file
 * convention, CLAUDE.md/coding-style.md) the same way task 8.1's
 * snapshot.c/snapshot_wire.c/snapshot_load.c split: shared, narrow helpers
 * are non-`static` here and redeclared via a plain `extern` at each
 * consumer's own use site, matching snapshot_wire.c's own documented idiom
 * rather than a shared private header.
 */
 /*
@ -169,3 +184,360 @@ int sched_validate_record_filename(const char* record_dir, const char* filename,
    return 0;
 }
 /* ========================================================================
 * Task 9.2: record start/stop lifecycle.
 * ======================================================================== */
 /*
 * write_capnp_message - serializes `c`'s current root message and appends
 * it to `f` as one self-framed Cap'n Proto message (Req 5 AC 20: "the
 * .replay file is a single stream of consecutive capnp messages ... no
 * extra outer framing needed"). Shared by every message writer in this
 * file and in recording_events.c/recording_session.c (non-`static`; see
 * this file's own header comment for why). Returns 0 on success; -EIO if
 * the message is empty or the full byte count could not be written;
 * -ENOMEM if the staging buffer could not be allocated.
 */
 int write_capnp_message(FILE* f, struct capn* c)
 {
    int64_t  sz;
    uint8_t* buf;
    size_t   written;
    sz = capn_size(c);
    if (sz <= 0) {
        return -EIO;
    }
    buf = (uint8_t*)malloc((size_t)sz);
    if (buf == NULL) {
        return -ENOMEM;
    }
    capn_write_mem(c, buf, (size_t)sz, 0);
    written = fwrite(buf, 1, (size_t)sz, f);
    free(buf);
    if (written != (size_t)sz) {
        return -EIO;
    }
    return 0;
 }
 /*
 * write_replay_header - builds and writes the ReplayHeader message (Req 5
 * AC 20's msg 1). `configHash` is intentionally left unset (reads back as
 * empty on decode): no utility anywhere in this codebase computes a config
 * hash today (verified -- the only sha256 handling elsewhere, e.g.
 * request.c's raw_sha256, is for CALLER-supplied bytes on the config-push
 * wire path, not something computed FROM an xce_conf4sched_t); introducing
 * one is out of this task's boundary. This mirrors the same documented
 * "Data field codec is a no-op" limitation typemap.h already calls out for
 * this exact field.
 */
 static int write_replay_header(FILE* f, xce_scheduler_t* sched,
                               uint64_t start_ts)
 {
    struct capn          c;
    struct capn_segment* cs;
    replay_header_t      hdr;
    replay_cluster_t     cluster;
    char**               controllers  = NULL;
    int                  ncontrollers = 0;
    char                 wall[32];
    time_t               now_wall;
    struct tm            tmv;
    ReplayHeader_ptr     ptr;
    int                  rc;
    int                  i;
    memset(&hdr, 0, sizeof(hdr));
    memset(&cluster, 0, sizeof(cluster));
    now_wall = time(NULL);
    gmtime_r(&now_wall, &tmv);
    strftime(wall, sizeof(wall), "%Y-%m-%dT%H:%M:%SZ", &tmv);
    hdr.recorder_version     = "1.0";
    hdr.recording_start_ts   = start_ts;
    hdr.recording_start_wall = wall;
    hdr.sched_policy         = sched->core.job_pick_plugin; /* may be NULL */
    ncontrollers = sched->cluster_cfg.candidates_count;
    if (ncontrollers > 0) {
        controllers = (char**)calloc((size_t)ncontrollers, sizeof(char*));
        if (controllers == NULL) {
            return -ENOMEM;
        }
        for (i = 0; i < ncontrollers; i++) {
            char hostbuf[XCE_MAXHOSTNAME_LEN + 16];
            snprintf(hostbuf, sizeof(hostbuf), "%s:%d",
                     sched->cluster_cfg.candidates[i].hostname,
                     sched->cluster_cfg.candidates[i].port);
            controllers[i] = strdup(hostbuf);
        }
        cluster.name           = sched->cluster_cfg.identity;
        cluster.controllers    = controllers;
        cluster.n_controllers  = ncontrollers;
        cluster.recording_host = sched->cluster_cfg.identity;
        hdr.cluster            = &cluster;
    }
    /* else: single-host mode (cluster_cfg.candidates_count == 0,
     * scheduler.h's own field comment) -- hdr.cluster stays NULL;
     * encode_ReplayHeader_ptr's transitive encode_ReplayCluster_ptr(...,
     * NULL) writes a null pointer, which decodes back as an empty
     * ReplayCluster, never a crash (same pattern every other optional
     * nested-struct field in this schema already relies on). */
    capn_init_malloc(&c);
    cs = capn_root(&c).seg;
    encode_ReplayHeader_ptr(cs, &ptr, &hdr);
    capn_setp(capn_root(&c), 0, ptr.p);
    rc = write_capnp_message(f, &c);
    capn_free(&c);
    if (controllers != NULL) {
        for (i = 0; i < ncontrollers; i++) {
            free(controllers[i]);
        }
        free(controllers);
    }
    return rc;
 }
 /*
 * write_snapshot_entry - builds and writes the initial SnapshotEntry
 * message (Req 5 AC 20's msg 2, "MUST be the first data message after
 * ReplayHeader"). Takes a full core snapshot via
 * xce_sched_core_save_snapshot() (core.h, task 8.1) and writes it.
 *
 * `snap`'s contents are deliberately never freed here: snapshot.c's own
 * "Ownership notes" (file header comment) document that `config` and each
 * `resources[]` entry are BORROWED live-core references, not deep copies
 * -- freeing them would corrupt/use-after-free the live scheduler. The
 * remaining fields (jobs/queues/workers/users/projects) ARE independently
 * heap-allocated by save_snapshot(), but no free-snapshot function exists
 * anywhere in this codebase yet (task 8.1's own test suite leaves every
 * save_snapshot() result unfreed for the same reason, detect_leaks=0
 * across this whole suite) -- writing one is out of this task's boundary.
 */
 static int write_snapshot_entry(FILE* f, xce_sched_core_t* core, uint64_t ts)
 {
    struct capn             c;
    struct capn_segment*    cs;
    sched_snapshot_data_t   snap;
    replay_snapshot_entry_t entry;
    SnapshotEntry_ptr       ptr;
    int                     rc;
    memset(&snap, 0, sizeof(snap));
    rc = xce_sched_core_save_snapshot(core, &snap);
    if (rc < 0) {
        return rc;
    }
    memset(&entry, 0, sizeof(entry));
    entry.ts   = ts;
    entry.data = &snap;
    capn_init_malloc(&c);
    cs = capn_root(&c).seg;
    encode_SnapshotEntry_ptr(cs, &ptr, &entry);
    capn_setp(capn_root(&c), 0, ptr.p);
    rc = write_capnp_message(f, &c);
    capn_free(&c);
    return rc;
 }
 int sched_record_start(xce_scheduler_t* sched, const char* record_file)
 {
    char        resolved_path[PATH_MAX];
    char        auto_name[64];
    const char* filename;
    int         fd;
    struct stat st;
    FILE*       f;
    uint64_t    start_ts;
    int         rc;
    if (sched == NULL) {
        return -EINVAL;
    }
    if (sched->recording) {
        return -EALREADY;
    }
    filename = record_file;
    if ((filename == NULL) || (filename[0] == '\0')) {
        char      timebuf[32];
        time_t    now_wall = time(NULL);
        struct tm tmv;
        gmtime_r(&now_wall, &tmv);
        strftime(timebuf, sizeof(timebuf), "%Y%m%d-%H%M%S", &tmv);
        snprintf(auto_name, sizeof(auto_name), "scheduler-replay-%s.replay",
                 timebuf);
        filename = auto_name;
    }
    rc = sched_validate_record_filename(sched->record_dir, filename,
                                        resolved_path, sizeof(resolved_path));
    if (rc < 0) {
        return rc;
    }
    /* TOCTOU hardening -- see recording.h's doc comment on
     * sched_validate_record_filename() and on sched_record_start() itself:
     * O_NOFOLLOW refuses a symlink swapped in after validation; the
     * fstat()+st_nlink==1 check below additionally refuses a hard link
     * swapped in (O_NOFOLLOW alone does not catch that -- no symlink is
     * involved in that race).
     *
     * Deliberately NO O_TRUNC here: O_TRUNC truncates atomically as PART
     * OF open() itself, before this function's own fstat()/st_nlink check
     * ever runs -- against a hard-linked victim file (same inode, no
     * symlink involved, so O_NOFOLLOW never triggers), that would corrupt
     * the victim the instant open() succeeds, regardless of what the
     * check below then decides. Instead: open non-destructively, verify
     * safety via fstat()+st_nlink, and only ftruncate() once verified --
     * so a rejected hard link is NEVER truncated. Matches
     * sched_validate_record_filename()'s own documented "legitimate
     * overwrite target" acceptance of a pre-existing single-link regular
     * file (which this ftruncate() then genuinely overwrites, same
     * end-visible behavior as O_TRUNC would have given for that safe
     * case). */
    fd = open(resolved_path, O_WRONLY | O_CREAT | O_NOFOLLOW, 0640);
    if (fd < 0) {
        return (errno > 0) ? -errno : -EIO;
    }
    if (fstat(fd, &st) != 0) {
        rc = (errno > 0) ? -errno : -EIO;
        close(fd);
        return rc;
    }
    if (!S_ISREG(st.st_mode) || (st.st_nlink != 1)) {
        close(fd);
        return -EACCES;
    }
    if (ftruncate(fd, 0) != 0) {
        rc = (errno > 0) ? -errno : -EIO;
        close(fd);
        return rc;
    }
    f = fdopen(fd, "w");
    if (f == NULL) {
        rc = (errno > 0) ? -errno : -EIO;
        close(fd);
        return rc;
    }
    start_ts = xce_core_now(&(sched->core));
    rc = write_replay_header(f, sched, start_ts);
    if (rc == 0) {
        rc = write_snapshot_entry(f, &(sched->core), start_ts);
    }
    if (rc < 0) {
        ERROR("%s failed to write replay header/snapshot to %s: rc=%d",
              __func__, resolved_path, rc);
        fclose(f);
        return rc;
    }
    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;
    return 0;
 }
 int sched_record_stop(xce_scheduler_t* sched)
 {
    struct capn          c;
    struct capn_segment* cs;
    replay_event_t       ev;
    ReplayEvent_ptr      ptr;
    int                  rc;
    if (sched == NULL) {
        return -EINVAL;
    }
    if (!sched->recording || (sched->record_file == NULL)) {
        return -EALREADY;
    }
    memset(&ev, 0, sizeof(ev));
    ev.ts         = xce_core_now(&(sched->core));
    ev.body_which = ReplayEvent_body_endSession;
    capn_init_malloc(&c);
    cs = capn_root(&c).seg;
    encode_ReplayEvent_ptr(cs, &ptr, &ev);
    capn_setp(capn_root(&c), 0, ptr.p);
    rc = write_capnp_message(sched->record_file, &c);
    capn_free(&c);
    if (rc < 0) {
        ERROR("%s failed to write end-of-recording marker to %s: rc=%d",
              __func__,
              (sched->record_file_path != NULL) ? sched->record_file_path : "?",
              rc);
    }
    /* Clear recording state unconditionally -- Req 5 AC 10's "Error
     * handling" clause: a failed write disables recording rather than
     * leaving it half-active. */
    fflush(sched->record_file);
    fclose(sched->record_file);
    sched->record_file = NULL;
    sched->recording   = false;
    free(sched->record_file_path);
    sched->record_file_path = NULL;
    return rc;
 }
 /*
 * sched_record_abort_on_error - shared "a recording write failed" cleanup
 * (Req 5 AC 10's "Error handling" clause), used by both
 * recording_events.c's sched_record_write_events() and
 * recording_session.c's sched_record_session_end() (non-`static`; see this
 * file's own header comment for why). Mirrors sched_record_stop()'s
 * cleanup, minus the end-of-recording marker write (the file is presumably
 * already broken -- disk full, permission revoked mid-write, etc. -- so
 * attempting one more write is not productive).
 */
 void sched_record_abort_on_error(xce_scheduler_t* sched, int rc)
 {
    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(sched->record_file);
        fclose(sched->record_file);
        sched->record_file = NULL;
    }
    free(sched->record_file_path);
    sched->record_file_path = NULL;
    sched->recording        = false;
 }
--- a/src/scheduler/src/recording_events.c
+++ b/src/scheduler/src/recording_events.c
@ -0,0 +1,177 @@
 #include <errno.h>
 #include <stdio.h>
 #include <string.h>
 #include "controller.capnp.h"
 #include "core.h"
 #include "recording.h"
 #include "utils.h"
 /*
 * recording_events.c - scheduler-core spec, task 9.2 (Req 5 AC 10, "On each
 * core_push_events() call ... For each event in the batch, serialize and
 * write the corresponding ReplayEvent message"). Split out of recording.c
 * (this project's 800-line-per-file convention, CLAUDE.md/coding-style.md)
 * -- see recording.c's own file header comment, "Split across three
 * files", for the full rationale. This file owns the per-event-batch
 * recording-write hook driver.c's xce_sched_driver_translate_pending_events()
 * calls into.
 *
 * write_capnp_message()/sched_record_abort_on_error() are defined
 * (non-`static`) in recording.c and redeclared here via a plain `extern`,
 * matching this codebase's existing cross-.c-file idiom for a narrow
 * shared dependency (snapshot_wire.c's own documented precedent, itself
 * citing core_session.c's `extern void sched_worker_release_host_ctx(...)`).
 */
 extern int  write_capnp_message(FILE* f, struct capn* c);
 extern void sched_record_abort_on_error(xce_scheduler_t* sched, int rc);
 /*
 * write_one_replay_event - builds and writes one ReplayEvent message from
 * a single translated `xce_core_event_t` (the SAME op-type table
 * driver.c's translate_one_pending_update() already established -- see
 * this function's per-case mapping below). Every wrapper local is declared
 * unconditionally (cheap stack space) so its address stays valid across
 * the single encode+write call after the switch, regardless of which case
 * populated it.
 */
 static int write_one_replay_event(FILE* f, xce_core_event_t* ev,
                                  uint64_t* event_count)
 {
    struct capn          c;
    struct capn_segment* cs;
    replay_event_t       wire;
    ReplayEvent_ptr      ptr;
    int                  rc;
    xce_jobs_data_t          jobs_wrap;
    xce_workers_data_t       workers_wrap;
    xce_resource_data_t      resources_wrap;
    xce_resval_list_t        resvals_wrap;
    job_state_change_event_t jsc_wrap;
    job_update_event_t       ju_wrap;
    memset(&wire, 0, sizeof(wire));
    memset(&jobs_wrap, 0, sizeof(jobs_wrap));
    memset(&workers_wrap, 0, sizeof(workers_wrap));
    memset(&resources_wrap, 0, sizeof(resources_wrap));
    memset(&resvals_wrap, 0, sizeof(resvals_wrap));
    memset(&jsc_wrap, 0, sizeof(jsc_wrap));
    memset(&ju_wrap, 0, sizeof(ju_wrap));
    wire.ts = ev->ts;
    switch (ev->type) {
    case XCE_CORE_EVENT_CONFIG_PUSH:
        wire.body_which       = ReplayEvent_body_configPush;
        wire.body.config_push = ev->u.config_push.config;
        break;
    case XCE_CORE_EVENT_JOBS_PUSH:
        jobs_wrap.n_total   = ev->u.jobs_push.njobs;
        jobs_wrap.count     = ev->u.jobs_push.njobs;
        jobs_wrap.jobs      = ev->u.jobs_push.jobs;
        wire.body_which     = ReplayEvent_body_jobsPush;
        wire.body.jobs_push = &jobs_wrap;
        break;
    case XCE_CORE_EVENT_WORKERS_PUSH:
        workers_wrap.n_total   = ev->u.workers_push.nworkers;
        workers_wrap.count     = ev->u.workers_push.nworkers;
        workers_wrap.workers   = ev->u.workers_push.workers;
        wire.body_which        = ReplayEvent_body_workersPush;
        wire.body.workers_push = &workers_wrap;
        break;
    case XCE_CORE_EVENT_RESOURCES_PUSH:
        resources_wrap.n_total   = ev->u.resources_push.n;
        resources_wrap.count     = ev->u.resources_push.n;
        resources_wrap.resources = ev->u.resources_push.resources;
        wire.body_which          = ReplayEvent_body_resourcesPush;
        wire.body.resources_push = &resources_wrap;
        break;
    case XCE_CORE_EVENT_RESVALS_FULL:
        resvals_wrap.count     = ev->u.resvals_full.n;
        resvals_wrap.vals      = ev->u.resvals_full.resvals;
        wire.body_which        = ReplayEvent_body_resvalsFull;
        wire.body.resvals_full = &resvals_wrap;
        break;
    case XCE_CORE_EVENT_RESVALS_DELTA:
        wire.body_which         = ReplayEvent_body_resvalsDelta;
        wire.body.resvals_delta = ev->u.resvals_delta.deltas;
        break;
    case XCE_CORE_EVENT_JOB_STATE_CHANGE:
        jsc_wrap.jobid             = ev->u.job_state_change.jobid;
        jsc_wrap.old_state         = ev->u.job_state_change.old_state;
        jsc_wrap.new_state         = ev->u.job_state_change.new_state;
        jsc_wrap.exit_code         = ev->u.job_state_change.exit_code;
        wire.body_which            = ReplayEvent_body_jobStateChange;
        wire.body.job_state_change = &jsc_wrap;
        break;
    case XCE_CORE_EVENT_JOB_UPDATE:
        ju_wrap.jobid          = ev->u.job_update.jobid;
        ju_wrap.changed_fields = ev->u.job_update.changed_fields;
        ju_wrap.job            = ev->u.job_update.job;
        wire.body_which        = ReplayEvent_body_jobUpdate;
        wire.body.job_update   = &ju_wrap;
        break;
    case XCE_CORE_EVENT_QUEUE_UPDATE:
        wire.body_which        = ReplayEvent_body_queueUpdate;
        wire.body.queue_update = ev->u.queue_update.queue;
        break;
    case XCE_CORE_EVENT_JOB_RUSAGE:
        wire.body_which      = ReplayEvent_body_jobRusage;
        wire.body.job_rusage = ev->u.job_rusage.rusages;
        break;
    default:
        WARN("%s unrecognized core event type %d, skipping recording write",
             __func__, (int)ev->type);
        return 0;
    }
    capn_init_malloc(&c);
    cs = capn_root(&c).seg;
    encode_ReplayEvent_ptr(cs, &ptr, &wire);
    capn_setp(capn_root(&c), 0, ptr.p);
    rc = write_capnp_message(f, &c);
    capn_free(&c);
    if (rc == 0) {
        (*event_count)++;
    }
    return rc;
 }
 int sched_record_write_events(xce_scheduler_t* sched, xce_core_event_t* events,
                              int n)
 {
    int i;
    if ((sched == NULL) || (events == NULL) || (n <= 0)) {
        return 0;
    }
    if (!sched->recording || (sched->record_file == NULL)) {
        return 0;
    }
    for (i = 0; i < n; i++) {
        int rc = write_one_replay_event(sched->record_file, &events[i],
                                        &(sched->record_event_count));
        if (rc < 0) {
            sched_record_abort_on_error(sched, rc);
            return rc;
        }
    }
    return 0;
 }
--- a/src/scheduler/src/recording_session.c
+++ b/src/scheduler/src/recording_session.c
@ -0,0 +1,244 @@
 #include <stdint.h>
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
 #include "controller.capnp.h"
 #include "core.h"
 #include "decision.h"
 #include "recording.h"
 #include "utils.h"
 /*
 * recording_session.c - scheduler-core spec, task 9.2 (Req 5 AC 10, "On
 * core_go(): begin building a ReplaySession message" / "In the decision
 * drain loop: append a ReplayDecision" / "On core_end_session(): finalize
 * ... write the complete message"). Split out of recording.c (this
 * project's 800-line-per-file convention, CLAUDE.md/coding-style.md) --
 * see recording.c's own file header comment, "Split across three files",
 * for the full rationale. This file owns the per-session ReplaySession
 * accumulator (sched_record_session_begin()/_add_decision()/_end())
 * driver.c's xce_sched_driver_run_normal_mode_round() calls into.
 *
 * write_capnp_message()/sched_record_abort_on_error() are defined
 * (non-`static`) in recording.c and redeclared here via a plain `extern`,
 * matching this codebase's existing cross-.c-file idiom for a narrow
 * shared dependency (snapshot_wire.c's own documented precedent, itself
 * citing core_session.c's `extern void sched_worker_release_host_ctx(...)`).
 */
 extern int  write_capnp_message(FILE* f, struct capn* c);
 extern void sched_record_abort_on_error(xce_scheduler_t* sched, int rc);
 /*
 * sched_record_session_ctx_s - see recording.h's forward-declared typedef
 * for the public (opaque) contract.
 */
 struct sched_record_session_ctx_s {
    uint64_t            begin_round;
    replay_decision_t** decisions;
    int                 n;
    int                 cap;
    uint32_t            n_dispatched;
    uint32_t            n_pended;
 };
 sched_record_session_ctx_t* sched_record_session_begin(xce_scheduler_t* sched,
                                                       uint64_t begin_round_ts)
 {
    sched_record_session_ctx_t* ctx;
    if ((sched == NULL) || !sched->recording || (sched->record_file == NULL)) {
        return NULL;
    }
    ctx = (sched_record_session_ctx_t*)calloc(1, sizeof(*ctx));
    if (ctx == NULL) {
        return NULL;
    }
    ctx->begin_round = begin_round_ts;
    return ctx;
 }
 /*
 * free_record_decision - frees one heap-allocated replay_decision_t built
 * by sched_record_session_add_decision() below, including the small
 * wrapper body struct it owns (dispatch/pend only -- preempt is never
 * appended, see that function's own comment). Never frees the underlying
 * xce_job_alloc_t / xce_job_preason_t payloads themselves -- those are
 * borrowed references into the live decision the driver drained, owned
 * elsewhere.
 */
 static void free_record_decision(replay_decision_t* rd)
 {
    if (rd == NULL) {
        return;
    }
    switch (rd->body_which) {
    case ReplayDecision_body_dispatch:
        free(rd->body.dispatch);
        break;
    case ReplayDecision_body_pend:
        free(rd->body.pend);
        break;
    default:
        break;
    }
    free(rd);
 }
 void sched_record_session_add_decision(sched_record_session_ctx_t* ctx,
                                       uint64_t ts, xce_decision_t* decision)
 {
    replay_decision_t* rd;
    if ((ctx == NULL) || (decision == NULL)) {
        return;
    }
    rd = (replay_decision_t*)calloc(1, sizeof(*rd));
    if (rd == NULL) {
        return;
    }
    rd->ts = ts;
    switch (decision->op) {
    case XCE_DECISION_DISPATCH: {
        sched_dispatch_decision_t* dd
            = (sched_dispatch_decision_t*)calloc(1, sizeof(*dd));
        if (dd == NULL) {
            free(rd);
            return;
        }
        dd->alloc         = decision->u.alloc;
        rd->body_which    = ReplayDecision_body_dispatch;
        rd->body.dispatch = dd;
        ctx->n_dispatched++;
        break;
    }
    case XCE_DECISION_PEND: {
        xce_pendreason_t*      pr = decision->u.reason;
        sched_pend_decision_t* pd;
        if (pr == NULL) {
            free(rd);
            return;
        }
        pd = (sched_pend_decision_t*)calloc(1, sizeof(*pd));
        if (pd == NULL) {
            free(rd);
            return;
        }
        pd->jobid      = pr->jobid;
        pd->reason     = pr->reasons;
        rd->body_which = ReplayDecision_body_pend;
        rd->body.pend  = pd;
        ctx->n_pended++;
        break;
    }
    case XCE_DECISION_PREEMPT:
        /* No live allocator path produces this decision type -- see
         * driver.c's xce_sched_driver_on_decision() header comment for the
         * full "documented unreachable, not silently worked around"
         * rationale, which this recording path mirrors: log loud, write
         * nothing to the recording for this decision, never fabricate wire
         * data for a shape this codebase cannot yet produce for real. */
        ERROR("%s PREEMPT decision recording not yet implemented "
              "(scheduler-preemption spec territory) -- jobid=%llu dropped "
              "from replay file",
              __func__,
              (unsigned long long)((decision->u.preempt != NULL)
                                       ? decision->u.preempt->jobid
                                       : 0));
        free(rd);
        return;
    default:
        /* XCE_DECISION_NONE / XCE_DECISION_NEW_SESSION: nothing to
         * record. */
        free(rd);
        return;
    }
    if (ctx->n == ctx->cap) {
        int                 newcap = ctx->cap ? (ctx->cap * 2) : 8;
        replay_decision_t** narr   = (replay_decision_t**)realloc(
            ctx->decisions, (size_t)newcap * sizeof(*narr));
        if (narr == NULL) {
            ERROR("%s out of memory growing decision batch (n=%d)", __func__,
                  ctx->n);
            free_record_decision(rd);
            return;
        }
        ctx->decisions = narr;
        ctx->cap       = newcap;
    }
    ctx->decisions[ctx->n++] = rd;
 }
 int sched_record_session_end(xce_scheduler_t*            sched,
                             sched_record_session_ctx_t* ctx, uint64_t end_ts)
 {
    struct capn          c;
    struct capn_segment* cs;
    replay_session_t     wire;
    session_summary_t    summary;
    ReplaySession_ptr    ptr;
    int                  rc;
    int                  i;
    if (ctx == NULL) {
        return 0;
    }
    if ((sched == NULL) || !sched->recording || (sched->record_file == NULL)) {
        for (i = 0; i < ctx->n; i++) {
            free_record_decision(ctx->decisions[i]);
        }
        free(ctx->decisions);
        free(ctx);
        return 0;
    }
    memset(&summary, 0, sizeof(summary));
    summary.duration_ns
        = (end_ts >= ctx->begin_round) ? (end_ts - ctx->begin_round) : 0;
    summary.n_decisions  = (uint32_t)ctx->n;
    summary.n_dispatched = ctx->n_dispatched;
    summary.n_pended     = ctx->n_pended;
    memset(&wire, 0, sizeof(wire));
    wire.begin_round = ctx->begin_round;
    wire.decisions   = ctx->decisions;
    wire.n_decisions = ctx->n;
    wire.summary     = &summary;
    capn_init_malloc(&c);
    cs = capn_root(&c).seg;
    encode_ReplaySession_ptr(cs, &ptr, &wire);
    capn_setp(capn_root(&c), 0, ptr.p);
    rc = write_capnp_message(sched->record_file, &c);
    capn_free(&c);
    if (rc < 0) {
        sched_record_abort_on_error(sched, rc);
    }
    for (i = 0; i < ctx->n; i++) {
        free_record_decision(ctx->decisions[i]);
    }
    free(ctx->decisions);
    free(ctx);
    return rc;
 }
--- a/src/scheduler/testsuite/meson.build
+++ b/src/scheduler/testsuite/meson.build
@ -412,6 +412,28 @@ sched_recording_path_validation_test = executable(
    dependencies: [scheduler_test_deps],
    install: false)
 # Contract test for the record start/stop lifecycle and the driver-loop
 # recording-write hooks (scheduler-core spec, task 9.2, Req 5 AC 10): start
 # writes a decodable ReplayHeader then SnapshotEntry; a full start -> submit
 # event batch -> complete one real session (real picker + stand-in
 # allocator threads, same rendezvous design as
 # sched_driver_session_ordering_test) -> stop cycle produces exactly the
 # expected message stream ending in ReplayEvent{endSession}; recording
 # inactive costs zero file I/O; double-start/stop-when-not-recording fail
 # cleanly; a TOCTOU-style symlink/hard-link swap between path validation
 # and the eventual open() is rejected. Same white-box #include pattern as
 # sched_driver_session_ordering_test (core.c/core_clock.c/core_events.c/
 # core_session.c/driver.c/plugin.c) PLUS ../src/recording.c/
 # recording_events.c/recording_session.c (the code under test) PLUS
 # ../src/snapshot.c/snapshot_wire.c (task 8.1's real
 # xce_sched_core_save_snapshot(), needed by sched_record_start()).
 sched_record_lifecycle_test = executable(
    'sched_record_lifecycle_test',
    ['test_record_lifecycle.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',
@ -475,3 +497,6 @@ test('scheduler snapshot save load test',
 test('scheduler recording path validation test',
     sched_recording_path_validation_test, suite: 'scheduler',
     env: ['ASAN_OPTIONS=detect_leaks=0'])
 test('scheduler record lifecycle test',
     sched_record_lifecycle_test, suite: 'scheduler',
     env: ['ASAN_OPTIONS=detect_leaks=0'])
--- a/src/scheduler/testsuite/test_driver_decision_cb.c
+++ b/src/scheduler/testsuite/test_driver_decision_cb.c
@ -115,6 +115,50 @@ int sched_account_jobs(xce_scheduler_t* sched)
    return 0;
 }
 /* Link-satisfying stubs for recording.c/recording_events.c/
 * recording_session.c's real functions -- as of scheduler-core spec task
 * 9.2, driver.c's (real, #included above)
 * xce_sched_driver_translate_pending_events()/
 * xce_sched_driver_run_normal_mode_round() reference these. Neither is
 * exercised by this file's tests (out of THIS file's scope -- see
 * test_driver_translate.c/test_driver_session_ordering.c), and this file
 * never links recording.c's real implementations (they need
 * xce_sched_core_save_snapshot(), snapshot.c). Never actually invoked at
 * runtime by any test in this file. */
 int sched_record_write_events(xce_scheduler_t* sched, xce_core_event_t* events,
                              int n)
 {
    (void)sched;
    (void)events;
    (void)n;
    return 0;
 }
 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 (this file never links
 * schedobj.c/scheduler.c). Never actually exercised (no test here pushes
--- a/src/scheduler/testsuite/test_driver_session_ordering.c
+++ b/src/scheduler/testsuite/test_driver_session_ordering.c
@ -201,6 +201,59 @@ int sched_fill_limit_requests(xce_scheduler_t* sched, xvec_idval_t* vec,
    return 0;
 }
 /* Link-satisfying stubs for recording.c/recording_events.c/
 * recording_session.c's real functions -- as of scheduler-core spec task
 * 9.2, driver.c's (real, #included above)
 * xce_sched_driver_translate_pending_events()/
 * xce_sched_driver_run_normal_mode_round() reference these
 * unconditionally (the `sched.recording`/`record_ctx != NULL` gates that
 * decide whether they do real work are runtime checks INSIDE these
 * functions, not something that removes the call site itself -- the
 * linker needs the symbol regardless). This file's fixture never sets
 * `sched.recording` (memset(&sched, 0, ...) below leaves it false), so at
 * runtime xce_sched_driver_translate_pending_events() calls this stub with
 * a non-NULL `sched` (record_ctx == &sched, always non-NULL here) but
 * xce_sched_driver_run_normal_mode_round() never calls
 * sched_record_session_begin() with `sched.recording == true` -- either
 * way, these stubs never need to do real work for this file's ordering
 * proof, which is not about recording. Never links recording.c's real
 * implementations (they need xce_sched_core_save_snapshot(), snapshot.c,
 * an even bigger dependency surface this file's own fixture doesn't
 * otherwise need). */
 int sched_record_write_events(xce_scheduler_t* sched, xce_core_event_t* events,
                              int n)
 {
    (void)sched;
    (void)events;
    (void)n;
    return 0;
 }
 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 -- see
 * file header comment (same rationale/algorithm as test_core_full_cycle.c's
 * own copy). */
--- a/src/scheduler/testsuite/test_driver_translate.c
+++ b/src/scheduler/testsuite/test_driver_translate.c
@ -90,6 +90,53 @@ int sched_publish_pend_reason(xce_scheduler_t* sched, xce_preason_set_t* r)
    return 0;
 }
 /* Link-satisfying stubs for recording.c/recording_events.c/
 * recording_session.c's real functions -- as of scheduler-core spec task
 * 9.2, driver.c's (real, #included above)
 * xce_sched_driver_translate_pending_events() calls sched_record_write_events()
 * unconditionally (behind a NULL `record_ctx` check, not a call this test ever
 * takes since every call site below passes NULL), and
 * xce_sched_driver_run_normal_mode_round() (out of THIS file's scope, same as
 * the core_session.c/plugin.c/ scheduler.c stubs above) calls the
 * sched_record_session_* trio. This file never links recording.c's real
 * implementations (they need xce_sched_core_save_snapshot(), snapshot.c, an
 * even bigger dependency surface this file's own header comment already avoids
 * pulling in for sched_publish_dispatch()/sched_publish_pend_reason()). Never
 * actually invoked at runtime by any test in this file. */
 int sched_record_write_events(xce_scheduler_t* sched, xce_core_event_t* events,
                              int n)
 {
    (void)sched;
    (void)events;
    (void)n;
    return 0;
 }
 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;
 }
 /* Link-satisfying stubs: as of scheduler-core spec task 5.2,
 * xce_sched_driver_run_normal_mode_round() (driver.c, real, #included
 * above, alongside xce_sched_driver_translate_pending_events() this file
@ -696,7 +743,7 @@ static void test_drain_translates_and_pushes_all_ten_op_types(void)
    r10->u.config                   = config;
    enqueue(&core, r10);
-    rc = xce_sched_driver_translate_pending_events(&core);
+    rc = xce_sched_driver_translate_pending_events(&core, NULL);
    TEST_ASSERT_EQUAL_INT(10, rc);
@ -746,7 +793,7 @@ static void test_accumulates_between_drains_and_empty_drain_is_noop(void)
    fixture_init(&core);
    /* Nothing pending yet. */
-    rc = xce_sched_driver_translate_pending_events(&core);
+    rc = xce_sched_driver_translate_pending_events(&core, NULL);
    TEST_ASSERT_EQUAL_INT(0, rc);
    /* One record enqueued -- drained on the next call. Heap-allocated (see
@ -764,13 +811,13 @@ static void test_accumulates_between_drains_and_empty_drain_is_noop(void)
        req->u.workerupdate.workers  = workers;
        enqueue(&core, req);
-        rc = xce_sched_driver_translate_pending_events(&core);
+        rc = xce_sched_driver_translate_pending_events(&core, NULL);
        TEST_ASSERT_EQUAL_INT(1, rc);
        TEST_ASSERT_NOT_NULL(sched_get_worker(&core, "hostB"));
    }
    /* Nothing new pending -- must not re-process anything. */
-    rc = xce_sched_driver_translate_pending_events(&core);
+    rc = xce_sched_driver_translate_pending_events(&core, NULL);
    TEST_ASSERT_EQUAL_INT(0, rc);
 }
@ -780,7 +827,7 @@ static void test_accumulates_between_drains_and_empty_drain_is_noop(void)
 static void test_null_core_returns_einval(void)
 {
-    int rc = xce_sched_driver_translate_pending_events(NULL);
+    int rc = xce_sched_driver_translate_pending_events(NULL, NULL);
    TEST_ASSERT_EQUAL_INT(-EINVAL, rc);
 }
@ -790,7 +837,7 @@ static void test_uninitialized_pend_updates_returns_einval(void)
    int              rc;
    memset(&core, 0, sizeof(core));
-    rc = xce_sched_driver_translate_pending_events(&core);
+    rc = xce_sched_driver_translate_pending_events(&core, NULL);
    TEST_ASSERT_EQUAL_INT(-EINVAL, rc);
 }
--- a/src/scheduler/testsuite/test_record_lifecycle.c
+++ b/src/scheduler/testsuite/test_record_lifecycle.c
@ -0,0 +1,826 @@
 /*
 * test_record_lifecycle.c - contract test for the record start/stop
 * lifecycle and the driver-loop recording-write hooks (scheduler-core
 * spec, task 9.2, Req 5 AC 10).
 *
 * Observable under test (this task's own text):
 *   - start -> the file begins with a decodable ReplayHeader then a
 *     decodable SnapshotEntry.
 *   - start -> submit an event batch -> complete one session -> stop
 *     produces a file with events and exactly one ReplaySession in
 *     between, ending in a ReplayEvent{endSession} marker.
 *   - when `sched->recording` is false throughout, no file I/O occurs.
 *   - double-start fails cleanly; stop-when-not-recording fails cleanly.
 *   - a TOCTOU-style symlink/hardlink swap between path validation and the
 *     eventual open() is rejected, not silently followed.
 *
 * White-box #include of ../src/core.c, ../src/core_clock.c,
 * ../src/core_events.c, ../src/core_session.c, ../src/driver.c,
 * ../src/plugin.c (same file set/rationale as
 * test_driver_session_ordering.c -- a real picker pthread + stand-in
 * allocator thread is needed to drive xce_sched_driver_run_normal_mode_round()
 * through one genuine session, jbuf's hardcoded capacity-1 rendezvous
 * requires it) PLUS ../src/recording.c, ../src/recording_events.c,
 * ../src/recording_session.c (the code under test) PLUS ../src/snapshot.c,
 * ../src/snapshot_wire.c (task 8.1's REAL xce_sched_core_save_snapshot(),
 * needed by sched_record_start() -- NOT snapshot_load.c, which nothing
 * here ever calls). No jobs are ever pushed (same "picker produces zero
 * decisions" fixture shape as test_driver_session_ordering.c), so
 * decision.c is not needed either.
 *
 * Decoding: this file reads its own produced `.replay` files back with the
 * SAME generated read_X()/decode_X() functions (controller.capnp.h) group
 * 7's own round-trip tests (test_e2e_capnp.c) use, via capn_init_fp()
 * (sequential per-message stream reads, since a `.replay` file is
 * consecutive self-framed capnp messages with no outer framing, Req 5
 * AC 20) -- never by hand-parsing bytes.
 *
 * Stubs below are the exact same set test_driver_session_ordering.c
 * already establishes for this core.c/core_clock.c/core_events.c/
 * core_session.c/driver.c/plugin.c combination (kill_sched_*,
 * sched_worker_release_host_ctx, sched_publish_dispatch/
 * sched_publish_pend_reason, sched_fill_limit_requests,
 * sched_resolve_job_picker's real reimplementation) -- see that file's own
 * header comment for the full rationale. plugin_unregister is NOT stubbed
 * (real, from the genuinely-linked plugin.c). No stub is needed for
 * xce_core_now()/xce_sched_core_save_snapshot() here (unlike
 * test_recording_path_validation.c's narrower TU): core_clock.c and
 * snapshot.c/snapshot_wire.c are genuinely linked into this file.
 */
 #include <errno.h>
 #include <fcntl.h>
 #include <limits.h>
 #include <pthread.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 "kvmgr.h"
 #include "preason.h"
 #include "scheduler.h"
 #include "xstring.h"
 /* Pull in the real functions under test into this TU (white-box
 * #include, same convention as test_driver_session_ordering.c). */
 #include "../src/core.c"
 #include "../src/core_clock.c"
 #include "../src/core_events.c"
 #include "../src/core_session.c"
 #include "../src/driver.c"
 #include "../src/plugin.c"
 #include "../src/recording.c"
 #include "../src/recording_events.c"
 #include "../src/recording_session.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; }
 /* core_reset_workers()'s (core_session.c) linker dependency -- never
 * invoked (no worker is ever added here). */
 void sched_worker_release_host_ctx(xce_sched_worker_t* w) { (void)w; }
 /* Link-satisfying stubs for scheduler.c's sched_publish_dispatch()/
 * sched_publish_pend_reason() -- never actually invoked (this fixture
 * pushes no jobs, so the driver's decision-drain loop never observes a
 * non-NULL decision). */
 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 stub for resource.c's sched_fill_limit_requests() --
 * plugin.c's plugin_prepare_allocation() references it; never actually
 * invoked here. */
 int sched_fill_limit_requests(xce_scheduler_t* sched, xvec_idval_t* vec,
                              xlist_t* limits)
 {
    (void)sched;
    (void)vec;
    (void)limits;
    return 0;
 }
 int sched_account_jobs(xce_scheduler_t* sched)
 {
    (void)sched;
    return 0;
 }
 /* Real reimplementation of schedobj.c's sched_resolve_job_picker -- same
 * rationale/algorithm as test_driver_session_ordering.c's own copy
 * (schedobj.c is not linked). */
 int sched_resolve_job_picker(xce_sched_core_t* core, xce_sched_plugin_t** out)
 {
    xce_sched_plugin_t* plugin = NULL;
    int                 rc     = 0;
    if ((core == NULL) || (out == NULL)) {
        return -EINVAL;
    }
    *out = NULL;
    if (core->job_pick_plugin == NULL) {
        return -EINVAL;
    }
    lock(core->plugins, RTTI_LOCK_READ);
    rc = key_find(core->plugins, core->job_pick_plugin, (void**)&plugin);
    unlock(core->plugins);
    if (rc <= 0) {
        return (rc < 0) ? rc : -ENOENT;
    }
    if ((plugin == NULL) || (plugin->picker_iface == NULL)
        || (plugin->picker_iface->next_job == NULL)
        || (plugin->picker_iface->begin_session == NULL)
        || (plugin->picker_iface->end_session == NULL)
        || (plugin->picker_iface->on_dispatch == NULL)
        || (plugin->picker_iface->on_pend == NULL)) {
        return -EINVAL;
    }
    *out = plugin;
    return 0;
 }
 void setUp(void) { return; }
 void tearDown(void) { return; }
 /* ---------------------------------------------------------------------- */
 /* Stub picker plugin -- identical shape to                                */
 /* test_driver_session_ordering.c's own: no jobs are ever pushed, so       */
 /* next_job() always returns NULL immediately. Its SOLE purpose is proving */
 /* the wbuf/jbuf rendezvous timing lets a real session complete.           */
 /* ---------------------------------------------------------------------- */
 static int stub_picker_begin_session(void* picker)
 {
    (void)picker;
    return 0;
 }
 static xce_sched_job_t* stub_picker_next_job(void* picker)
 {
    (void)picker;
    return NULL;
 }
 static int stub_picker_end_session(void* picker)
 {
    (void)picker;
    return 0;
 }
 static int stub_picker_jobid_noop(void* picker, uint64_t jobid)
 {
    (void)picker;
    (void)jobid;
    return 0;
 }
 static int stub_picker_queue_update_noop(void* picker, xce_sched_queue_t* q)
 {
    (void)picker;
    (void)q;
    return 0;
 }
 static int stub_picker_job_noop(void* picker, xce_sched_job_t* job)
 {
    (void)picker;
    (void)job;
    return 0;
 }
 static int stub_picker_job_state_change_noop(void* picker, xce_sched_job_t* job,
                                             uint32_t state)
 {
    (void)picker;
    (void)job;
    (void)state;
    return 0;
 }
 static xce_job_picker_iface_t g_stub_picker_iface
    = { .init                = NULL,
        .fini                = NULL,
        .begin_session       = stub_picker_begin_session,
        .next_job            = stub_picker_next_job,
        .end_session         = stub_picker_end_session,
        .on_dispatch         = stub_picker_jobid_noop,
        .on_pend             = stub_picker_jobid_noop,
        .on_queue_update     = stub_picker_queue_update_noop,
        .on_job_push         = stub_picker_job_noop,
        .on_job_state_change = stub_picker_job_state_change_noop,
        .on_job_update       = stub_picker_job_noop,
        .on_config           = NULL,
        .on_job_rusage       = NULL };
 static xce_sched_plugin_t g_stub_picker_plugin
    = { .name         = "test_record_lifecycle_picker",
        .picker_iface = &g_stub_picker_iface };
 static void* picker_thread_fn(void* arg)
 {
    xce_sched_core_t*   core       = (xce_sched_core_t*)arg;
    xce_sched_plugin_t* job_picker = NULL;
    int                 rc;
    int                 go = 0;
    uint64_t            jobid;
    rc = sched_resolve_job_picker(core, &job_picker);
    if (rc < 0) {
        return NULL;
    }
    cbuf_get(&(core->wbuf), &go);
    job_picker->picker_iface->begin_session(job_picker);
    for (;;) {
        xce_sched_job_t* job = job_picker->picker_iface->next_job(job_picker);
        if (job == NULL) {
            break;
        }
        jobid = job->jobid;
        cbuf_put(&(core->jbuf), &jobid);
    }
    jobid = 0;
    cbuf_put(&(core->jbuf), &jobid);
    job_picker->picker_iface->end_session(job_picker);
    return NULL;
 }
 static void* stand_in_allocator_thread_fn(void* arg)
 {
    xce_sched_core_t* core = (xce_sched_core_t*)arg;
    uint64_t          jobid;
    xce_decision_t*   decision = NULL;
    cbuf_get(&(core->jbuf), &jobid);
    (void)jobid;
    cbuf_put(&(core->decisions), &decision);
    return NULL;
 }
 /* ---------------------------------------------------------------------- */
 /* Fixture helpers                                                         */
 /* ---------------------------------------------------------------------- */
 static void make_tmp_dir(char* out, size_t out_len, const char* suffix)
 {
    char tmpl[PATH_MAX];
    snprintf(tmpl, sizeof(tmpl), "/tmp/sched_reclife%s_XXXXXX", suffix);
    TEST_ASSERT_NOT_NULL(mkdtemp(tmpl));
    TEST_ASSERT_TRUE(strlen(tmpl) < out_len);
    strcpy(out, tmpl);
 }
 /* Real xce_sched_core_init() + manually dataq_init'd pend_updates + the
 * stub picker plugin registered -- same pattern as
 * test_driver_session_ordering.c's fixture. `record_dir` must outlive
 * `sched` (caller-owned storage). Non-const to match
 * xce_scheduler_t.record_dir's own `char*` type (scheduler.h) -- every
 * caller passes a mutable local buffer anyway. */
 static void fixture_init(xce_scheduler_t* sched, char* record_dir)
 {
    int rc;
    memset(sched, 0, sizeof(*sched));
    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);
    lock(sched->core.plugins, RTTI_LOCK_WRITE);
    rc = key_add(sched->core.plugins, (void*)g_stub_picker_plugin.name,
                 (void*)&g_stub_picker_plugin);
    TEST_ASSERT_GREATER_OR_EQUAL(0, rc);
    unlock(sched->core.plugins);
    sched->core.job_pick_plugin = g_stub_picker_plugin.name;
    sched->record_dir           = record_dir;
 }
 /* Runs one complete normal-mode round (real picker thread + stand-in
 * allocator thread, same design as
 * test_driver_session_ordering.c's own single test) and blocks until it
 * completes. */
 static void run_one_real_session(xce_scheduler_t* sched)
 {
    pthread_t picker_tid;
    pthread_t allocator_tid;
    int       rc;
    rc = pthread_create(&picker_tid, NULL, picker_thread_fn, &(sched->core));
    TEST_ASSERT_EQUAL_INT(0, rc);
    rc = pthread_create(&allocator_tid, NULL, stand_in_allocator_thread_fn,
                        &(sched->core));
    TEST_ASSERT_EQUAL_INT(0, rc);
    xce_sched_driver_run_normal_mode_round(sched);
    rc = pthread_join(picker_tid, NULL);
    TEST_ASSERT_EQUAL_INT(0, rc);
    rc = pthread_join(allocator_tid, NULL);
    TEST_ASSERT_EQUAL_INT(0, rc);
 }
 /* ---------------------------------------------------------------------- */
 /* Test 1: start() writes a decodable ReplayHeader then a decodable        */
 /* SnapshotEntry -- this task's own Observable, "byte-decodable by the     */
 /* round-trip decoder from group 7".                                       */
 /* ---------------------------------------------------------------------- */
 static void test_start_writes_decodable_header_then_snapshot_entry(void)
 {
    char              record_dir[PATH_MAX];
    xce_scheduler_t   sched;
    char              path_copy[PATH_MAX];
    int               rc;
    FILE*             f;
    struct capn       c1;
    struct capn       c2;
    ReplayHeader_ptr  hdr;
    SnapshotEntry_ptr snap;
    capn_text         rv;
    make_tmp_dir(record_dir, sizeof(record_dir), "_hdr");
    fixture_init(&sched, record_dir);
    rc = sched_record_start(&sched, "session.replay");
    TEST_ASSERT_EQUAL_INT(0, rc);
    TEST_ASSERT_TRUE(sched.recording);
    TEST_ASSERT_NOT_NULL(sched.record_file_path);
    TEST_ASSERT_NOT_NULL(sched.record_file);
    strcpy(path_copy, sched.record_file_path);
    rc = sched_record_stop(&sched);
    TEST_ASSERT_EQUAL_INT(0, rc);
    TEST_ASSERT_FALSE(sched.recording);
    f = fopen(path_copy, "r");
    TEST_ASSERT_NOT_NULL(f);
    /* msg 1: ReplayHeader */
    TEST_ASSERT_EQUAL_INT(0, capn_init_fp(&c1, f, 0));
    hdr.p = capn_getp(capn_root(&c1), 0, 1);
    rv    = ReplayHeader_get_recorderVersion(hdr);
    TEST_ASSERT_EQUAL_STRING("1.0", rv.str);
    capn_free(&c1);
    /* msg 2: SnapshotEntry -- MUST be the first data message after
     * ReplayHeader (Req 5 AC 20). */
    TEST_ASSERT_EQUAL_INT(0, capn_init_fp(&c2, f, 0));
    snap.p = capn_getp(capn_root(&c2), 0, 1);
    {
        SnapshotData_ptr sdata = SnapshotEntry_get_data(snap);
        TEST_ASSERT_FALSE(sdata.p.type == CAPN_NULL);
    }
    capn_free(&c2);
    fclose(f);
    printf("test_start_writes_decodable_header_then_snapshot_entry: PASS\n");
 }
 /* ---------------------------------------------------------------------- */
 /* Test 2: start -> submit an event batch -> complete one session -> stop  */
 /* produces events + exactly one ReplaySession + a trailing endSession     */
 /* marker -- this task's own literal Observable.                          */
 /* ---------------------------------------------------------------------- */
 static void
 test_full_cycle_start_event_session_stop_produces_expected_stream(void)
 {
    char                        record_dir[PATH_MAX];
    xce_scheduler_t             sched;
    xce_sched_update_request_t* req;
    char                        path_copy[PATH_MAX];
    int                         rc;
    FILE*                       f;
    struct capn                 c;
    make_tmp_dir(record_dir, sizeof(record_dir), "_cycle");
    fixture_init(&sched, record_dir);
    rc = sched_record_start(&sched, "cycle.replay");
    TEST_ASSERT_EQUAL_INT(0, rc);
    strcpy(path_copy, sched.record_file_path);
    /* "submit an event batch": one JSTATE_CHANGE update, drained by
     * xce_sched_driver_translate_pending_events() (called from
     * xce_sched_driver_run_normal_mode_round() below) -- the SAME
     * pend_updates dataq mechanism handler.c populates in production. */
    req     = (xce_sched_update_request_t*)calloc(1, sizeof(*req));
    req->op = XCE_SCHED_OP_JSTATE_CHANGE;
    req->u.job_state_change.jobid = 42ULL;
    req->u.job_state_change.state = 7u;
    dataq_enq(sched.core.pend_updates[0], &(req->link));
    /* "complete a session": one full normal-mode round through a real
     * picker thread + stand-in allocator thread (no jobs pushed, so the
     * session produces zero decisions -- still exactly one ReplaySession
     * message, matching this task's Observable which requires "one
     * session in between", not any particular decision count). */
    run_one_real_session(&sched);
    TEST_ASSERT_EQUAL_UINT64(1ULL, sched.record_event_count);
    rc = sched_record_stop(&sched);
    TEST_ASSERT_EQUAL_INT(0, rc);
    f = fopen(path_copy, "r");
    TEST_ASSERT_NOT_NULL(f);
    /* msg 1: ReplayHeader */
    TEST_ASSERT_EQUAL_INT(0, capn_init_fp(&c, f, 0));
    capn_free(&c);
    /* msg 2: SnapshotEntry (initial) */
    TEST_ASSERT_EQUAL_INT(0, capn_init_fp(&c, f, 0));
    capn_free(&c);
    /* msg 3: ReplayEvent{jobStateChange, jobid=42} -- the submitted batch. */
    {
        ReplayEvent_ptr    p;
        struct ReplayEvent wire;
        replay_event_t     dst;
        TEST_ASSERT_EQUAL_INT(0, capn_init_fp(&c, f, 0));
        p.p = capn_getp(capn_root(&c), 0, 1);
        read_ReplayEvent(&wire, p);
        memset(&dst, 0, sizeof(dst));
        decode_ReplayEvent(&dst, &wire);
        TEST_ASSERT_EQUAL_INT((int)ReplayEvent_body_jobStateChange,
                              (int)dst.body_which);
        TEST_ASSERT_NOT_NULL(dst.body.job_state_change);
        TEST_ASSERT_EQUAL_UINT64(42ULL, dst.body.job_state_change->jobid);
        free_ReplayEvent(&dst);
        capn_free(&c);
    }
    /* msg 4: exactly one ReplaySession. */
    {
        ReplaySession_ptr    p;
        struct ReplaySession wire;
        TEST_ASSERT_EQUAL_INT(0, capn_init_fp(&c, f, 0));
        p.p = capn_getp(capn_root(&c), 0, 1);
        read_ReplaySession(&wire, p);
        (void)wire; /* begin_round/decisions checked structurally only --
                     * this fixture pushes no jobs, so 0 decisions is
                     * correct and expected. */
        capn_free(&c);
    }
    /* msg 5: the end-of-recording marker, ReplayEvent{endSession}. */
    {
        ReplayEvent_ptr    p;
        struct ReplayEvent wire;
        replay_event_t     dst;
        TEST_ASSERT_EQUAL_INT(0, capn_init_fp(&c, f, 0));
        p.p = capn_getp(capn_root(&c), 0, 1);
        read_ReplayEvent(&wire, p);
        memset(&dst, 0, sizeof(dst));
        decode_ReplayEvent(&dst, &wire);
        TEST_ASSERT_EQUAL_INT((int)ReplayEvent_body_endSession,
                              (int)dst.body_which);
        free_ReplayEvent(&dst);
        capn_free(&c);
    }
    /* Exactly 5 messages -- nothing more to read. */
    TEST_ASSERT_NOT_EQUAL(0, capn_init_fp(&c, f, 0));
    fclose(f);
    printf("test_full_cycle_start_event_session_stop_produces_expected_stream:"
           " PASS\n");
 }
 /* ---------------------------------------------------------------------- */
 /* Test 3: when recording is inactive, the driver-loop hooks touch no      */
 /* file/record state -- zero file I/O (Req 5 AC 10, "Performance").        */
 /* ---------------------------------------------------------------------- */
 static void test_recording_inactive_hooks_are_zero_cost_noops(void)
 {
    char                        record_dir[PATH_MAX];
    xce_scheduler_t             sched;
    xce_core_event_t            events[1];
    sched_record_session_ctx_t* ctx;
    int                         rc;
    make_tmp_dir(record_dir, sizeof(record_dir), "_inactive");
    fixture_init(&sched, record_dir);
    /* sched.recording deliberately never set true. */
    memset(events, 0, sizeof(events));
    events[0].type                     = XCE_CORE_EVENT_JOB_STATE_CHANGE;
    events[0].u.job_state_change.jobid = 99ULL;
    rc = sched_record_write_events(&sched, events, 1);
    TEST_ASSERT_EQUAL_INT(0, rc);
    TEST_ASSERT_NULL(sched.record_file);
    TEST_ASSERT_NULL(sched.record_file_path);
    ctx = sched_record_session_begin(&sched, 123ULL);
    TEST_ASSERT_NULL(ctx);
    TEST_ASSERT_NULL(sched.record_file);
    /* Driving a full real round with recording inactive must not create
     * any file either. */
    run_one_real_session(&sched);
    TEST_ASSERT_FALSE(sched.recording);
    TEST_ASSERT_NULL(sched.record_file);
    TEST_ASSERT_NULL(sched.record_file_path);
    TEST_ASSERT_EQUAL_UINT64(0ULL, sched.record_event_count);
 }
 /* ---------------------------------------------------------------------- */
 /* Test 4: double-start / stop-when-not-recording fail cleanly.            */
 /* ---------------------------------------------------------------------- */
 static void test_double_start_fails_cleanly(void)
 {
    char            record_dir[PATH_MAX];
    xce_scheduler_t sched;
    int             rc;
    make_tmp_dir(record_dir, sizeof(record_dir), "_dblstart");
    fixture_init(&sched, record_dir);
    rc = sched_record_start(&sched, "a.replay");
    TEST_ASSERT_EQUAL_INT(0, rc);
    rc = sched_record_start(&sched, "b.replay");
    TEST_ASSERT_EQUAL_INT(-EALREADY, rc);
    /* Original recording untouched by the rejected second start. */
    TEST_ASSERT_TRUE(sched.recording);
    TEST_ASSERT_TRUE(strstr(sched.record_file_path, "a.replay") != NULL);
    rc = sched_record_stop(&sched);
    TEST_ASSERT_EQUAL_INT(0, rc);
 }
 static void test_stop_when_not_recording_fails_cleanly(void)
 {
    char            record_dir[PATH_MAX];
    xce_scheduler_t sched;
    int             rc;
    make_tmp_dir(record_dir, sizeof(record_dir), "_nostart");
    fixture_init(&sched, record_dir);
    rc = sched_record_stop(&sched);
    TEST_ASSERT_EQUAL_INT(-EALREADY, rc);
    rc = sched_record_start(&sched, "c.replay");
    TEST_ASSERT_EQUAL_INT(0, rc);
    rc = sched_record_stop(&sched);
    TEST_ASSERT_EQUAL_INT(0, rc);
    /* Stopping again after a clean stop also fails cleanly. */
    rc = sched_record_stop(&sched);
    TEST_ASSERT_EQUAL_INT(-EALREADY, rc);
 }
 static void test_null_sched_returns_einval(void)
 {
    TEST_ASSERT_EQUAL_INT(-EINVAL, sched_record_start(NULL, "x.replay"));
    TEST_ASSERT_EQUAL_INT(-EINVAL, sched_record_stop(NULL));
 }
 /* ---------------------------------------------------------------------- */
 /* Test 5: TOCTOU -- a symlink/hard link swapped in at the resolved path   */
 /* between path validation and the eventual open() is rejected, not        */
 /* silently followed (recording.h's own documented mandate for this task). */
 /*                                                                          */
 /* Genuinely racing a background thread against the exact few-instruction  */
 /* window inside sched_record_start() between its internal                */
 /* sched_validate_record_filename() call and its open() call would be      */
 /* inherently flaky (no test hook exists to pause mid-function, nor should */
 /* one be added to production code just for this). Instead, both tests     */
 /* below reproduce the STATE a genuine race would produce -- a malicious   */
 /* symlink/hard link occupying the target path at the moment              */
 /* sched_record_start() is invoked, planted via real symlink(2)/link(2)    */
 /* syscalls (no mocking) -- and prove the full end-to-end call rejects it, */
 /* AND (the part that specifically proves the open()-time hardening this   */
 /* task added, not just sched_validate_record_filename()'s pre-existing    */
 /* containment check) that the exact open()+fstat()+st_nlink==1 sequence   */
 /* sched_record_start() runs internally -- reproduced here verbatim, same  */
 /* flags -- independently rejects the swap. Either layer alone would be    */
 /* enough in production; proving the second layer here is what closes the  */
 /* TOCTOU gap this task's own docstring (recording.h) flags: a real race   */
 /* would land AFTER sched_validate_record_filename() already accepted the  */
 /* path, i.e. exactly the scenario where ONLY the open()-time check saves  */
 /* it.                                                                     */
 /* ---------------------------------------------------------------------- */
 static int toctou_hardened_open(const char* path)
 {
    int         fd;
    struct stat st;
    /* Verbatim copy of sched_record_start()'s own hardening sequence
     * (recording.c) -- see this test's own header comment for why this is
     * reproduced rather than racing a real thread. Deliberately NO
     * O_TRUNC: it truncates atomically as part of open() itself, before
     * the fstat()/st_nlink check below ever runs -- against a hard-linked
     * victim (same inode, no symlink involved) that would corrupt the
     * victim regardless of what the check then decides. ftruncate() only
     * runs after the check passes. */
    fd = open(path, O_WRONLY | O_CREAT | O_NOFOLLOW, 0640);
    if (fd < 0) {
        return (errno > 0) ? -errno : -EIO;
    }
    if (fstat(fd, &st) != 0) {
        int rc = (errno > 0) ? -errno : -EIO;
        close(fd);
        return rc;
    }
    if (!S_ISREG(st.st_mode) || (st.st_nlink != 1)) {
        close(fd);
        return -EACCES;
    }
    if (ftruncate(fd, 0) != 0) {
        int rc = (errno > 0) ? -errno : -EIO;
        close(fd);
        return rc;
    }
    close(fd);
    return 0;
 }
 static void test_toctou_symlink_planted_after_validation_rejected(void)
 {
    char        record_dir[PATH_MAX];
    char        outside_dir[PATH_MAX];
    char        outside_secret[PATH_MAX];
    char        resolved[PATH_MAX];
    char        target[PATH_MAX];
    const char* secret_content = "do-not-touch";
    int         rc;
    FILE*       f;
    char        readback[64];
    make_tmp_dir(record_dir, sizeof(record_dir), "_toctou_sym_src");
    make_tmp_dir(outside_dir, sizeof(outside_dir), "_toctou_sym_dst");
    snprintf(outside_secret, sizeof(outside_secret), "%s/secret", outside_dir);
    f = fopen(outside_secret, "w");
    TEST_ASSERT_NOT_NULL(f);
    fputs(secret_content, f);
    fclose(f);
    /* Simulates "validation just resolved this path and found nothing
     * there yet" (a fresh filename -- sched_validate_record_filename()
     * itself accepts this, since lstat() on a nonexistent entry short-
     * circuits its containment checks entirely). */
    rc = sched_validate_record_filename(record_dir, "race.replay", resolved,
                                        sizeof(resolved));
    TEST_ASSERT_EQUAL_INT(0, rc);
    /* The race: a symlink appears at the just-validated path, pointing
     * OUTSIDE record_dir, in the window before open() runs. */
    snprintf(target, sizeof(target), "%s/race.replay", record_dir);
    TEST_ASSERT_EQUAL_INT(0, symlink(outside_secret, target));
    /* Layer 1: the open()-time hardening sequence itself rejects it. */
    rc = toctou_hardened_open(resolved);
    TEST_ASSERT_TRUE(rc < 0);
    /* Layer 2 (defense in depth): a real end-to-end sched_record_start()
     * call against the same record_dir/filename is ALSO rejected --
     * sched_validate_record_filename() re-runs and now sees the symlink,
     * closing the loop even though this particular ordering happens to be
     * caught one layer earlier than the pure race would be. */
    {
        xce_scheduler_t sched;
        fixture_init(&sched, record_dir);
        rc = sched_record_start(&sched, "race.replay");
        TEST_ASSERT_TRUE(rc < 0);
        TEST_ASSERT_FALSE(sched.recording);
        TEST_ASSERT_NULL(sched.record_file);
    }
    /* The outside secret file was never opened for writing/truncated. */
    f = fopen(outside_secret, "r");
    TEST_ASSERT_NOT_NULL(f);
    memset(readback, 0, sizeof(readback));
    TEST_ASSERT_NOT_NULL(fgets(readback, sizeof(readback), f));
    fclose(f);
    TEST_ASSERT_EQUAL_STRING(secret_content, readback);
 }
 static void test_toctou_hardlink_planted_after_validation_rejected(void)
 {
    char        record_dir[PATH_MAX];
    char        outside_dir[PATH_MAX];
    char        victim[PATH_MAX];
    char        resolved[PATH_MAX];
    char        aliased[PATH_MAX];
    const char* victim_content = "do-not-alias";
    int         rc;
    FILE*       f;
    char        readback[64];
    make_tmp_dir(record_dir, sizeof(record_dir), "_toctou_hl_src");
    make_tmp_dir(outside_dir, sizeof(outside_dir), "_toctou_hl_dst");
    snprintf(victim, sizeof(victim), "%s/victim", outside_dir);
    f = fopen(victim, "w");
    TEST_ASSERT_NOT_NULL(f);
    fputs(victim_content, f);
    fclose(f);
    rc = sched_validate_record_filename(record_dir, "alias.replay", resolved,
                                        sizeof(resolved));
    TEST_ASSERT_EQUAL_INT(0, rc);
    /* The race: a hard link to the OUTSIDE victim file appears at the
     * just-validated path -- same inode, no symlink involved, so
     * O_NOFOLLOW alone would NOT catch this (recording.h's own doc
     * comment: "no symlink is involved in that race, so O_NOFOLLOW never
     * triggers") -- only the fstat()+st_nlink==1 check does. */
    snprintf(aliased, sizeof(aliased), "%s/alias.replay", record_dir);
    TEST_ASSERT_EQUAL_INT(0, link(victim, aliased));
    rc = toctou_hardened_open(resolved);
    TEST_ASSERT_EQUAL_INT(-EACCES, rc);
    {
        xce_scheduler_t sched;
        fixture_init(&sched, record_dir);
        rc = sched_record_start(&sched, "alias.replay");
        TEST_ASSERT_TRUE(rc < 0);
        TEST_ASSERT_FALSE(sched.recording);
        TEST_ASSERT_NULL(sched.record_file);
    }
    f = fopen(victim, "r");
    TEST_ASSERT_NOT_NULL(f);
    memset(readback, 0, sizeof(readback));
    TEST_ASSERT_NOT_NULL(fgets(readback, sizeof(readback), f));
    fclose(f);
    TEST_ASSERT_EQUAL_STRING(victim_content, readback);
 }
 int main(void)
 {
    UNITY_BEGIN();
    RUN_TEST(test_start_writes_decodable_header_then_snapshot_entry);
    RUN_TEST(test_full_cycle_start_event_session_stop_produces_expected_stream);
    RUN_TEST(test_recording_inactive_hooks_are_zero_cost_noops);
    RUN_TEST(test_double_start_fails_cleanly);
    RUN_TEST(test_stop_when_not_recording_fails_cleanly);
    RUN_TEST(test_null_sched_returns_einval);
    RUN_TEST(test_toctou_symlink_planted_after_validation_rejected);
    RUN_TEST(test_toctou_hardlink_planted_after_validation_rejected);
    return UNITY_END();
 }
--- a/src/scheduler/testsuite/test_recording_path_validation.c
+++ b/src/scheduler/testsuite/test_recording_path_validation.c
@ -43,6 +43,30 @@
 /* Pull in the real sched_validate_record_filename() under test. */
 #include "../src/recording.c"
 /* Link-satisfying stubs: recording.c also defines sched_record_start()/
 * sched_record_stop() (scheduler-core spec, task 9.2) as of this revision,
 * which reference xce_core_now() (core_clock.c) and
 * xce_sched_core_save_snapshot() (snapshot.c) -- neither pulled into this
 * TU, same rationale as every other narrow white-box test in this
 * directory (this file's own header comment: "the function under test has
 * no internal state or fixture dependencies", still true of
 * sched_validate_record_filename() itself; this task's own additions are
 * exercised by a separate, dedicated test file). Never actually invoked at
 * runtime by any test in this file. */
 uint64_t xce_core_now(xce_sched_core_t* core)
 {
    (void)core;
    return 0;
 }
 int xce_sched_core_save_snapshot(xce_sched_core_t*      core,
                                 sched_snapshot_data_t* out)
 {
    (void)core;
    (void)out;
    return -ENOSYS;
 }
 void setUp(void) { return; }
 void tearDown(void) { return; }