From 68a6196517b8d21816af1df8e5c0663cde2d7073 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Tue, 4 Aug 2026 09:38:13 +0000 Subject: [PATCH] feat(scheduler-core): define replay stream envelope schema (task 7.1) Adds ReplayCluster/ReplayHeader/SnapshotEntry/ReplayEvent/ SessionSummary/ReplaySession to controller.capnp, additively, first task of Phase 3 (Foundation: Replay Wire Format). Requirements.md's AC 20 capnp sketch isn't self-contained -- ReplayEvent's union references JobStateChangeEvent/JobUpdateEvent (task 7.2's job), SnapshotEntry.data/ReplayDecision's union reference SnapshotData/ DispatchDecision/PendDecision/PreemptDecision (task 7.3's job), none of which exist yet. Resolved by verifying capnp's real constraints against the toolchain directly: SnapshotData is an empty {} placeholder (task 7.3 replaces its mapname'd C struct wholesale when adding the real ~11 fields), ReplayDecision holds only {ts} (task 7.3 appends a body union at new ordinals), and ReplayEvent's union holds only the 9 members with pre-existing backing types (Conf4Sched/SchedJobData/ WorkerData/ResourceData/ResValList/QueueConf/SchedJobRusages/Void), deferring jobStateChange/jobUpdate to task 7.2 as new ordinals -- capnp requires strictly sequential ordinals with no holes, so this split is compiler-mandated, not stylistic. Round-trip test added to test_e2e_capnp.c. Full project build (all binaries sharing this schema) and 1075/1075 controller_unit_test verified clean. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_019dw5WnN3DXStaEaPYfACU1 --- .kiro/specs/scheduler-core/tasks.md | 3 +- src/controller/testsuite/test_e2e_capnp.c | 216 ++++++++++++++++++++++ src/include/typemap.h | 84 +++++++++ src/schema/controller.capnp | 105 +++++++++++ 4 files changed, 407 insertions(+), 1 deletion(-) diff --git a/.kiro/specs/scheduler-core/tasks.md b/.kiro/specs/scheduler-core/tasks.md index bad4f7a0..f4368ade 100644 --- a/.kiro/specs/scheduler-core/tasks.md +++ b/.kiro/specs/scheduler-core/tasks.md @@ -158,10 +158,11 @@ - [ ] 7. Foundation: Replay Wire Format -- [ ] 7.1 Define the replay stream envelope schema +- [x] 7.1 Define the replay stream envelope schema - Add the replay header, snapshot entry, replay event, replay session, and replay decision wrapper message types to the shared wire schema, following the single-stream-of-consecutive-messages framing - Observable: the schema compiles and a minimal header-plus-snapshot-entry byte stream round-trips through encode/decode - _Requirements: 5.20_ + - _Verified: 2026-08-04 — ReplayCluster/ReplayHeader/SnapshotEntry/ReplayEvent/SessionSummary/ReplaySession added fully; SnapshotData (empty {}) and ReplayDecision ({ts only}) added as compiler-mandated placeholders task 7.3 fills in (real fields for SnapshotData wholesale-replaced, a body union appended to ReplayDecision at new ordinals) since capnp's $C.codecgen has no per-struct opt-out from requiring the mapname'd C type to exist. ReplayEvent's union deliberately holds only the 9 members with pre-existing backing types (configPush/jobsPush/workersPush/resourcesPush/resvalsFull/resvalsDelta/queueUpdate/jobRusage/endSession); jobStateChange/jobUpdate deferred to task 7.2 as new ordinals — proven compiler-mandatory (not stylistic) by review, which independently confirmed capnp rejects both the AC 20 pseudocode's literal union syntax and any ordinal gap. Round-trip test (test_e2e_capnp.c) proven real via full clean rebuild (deleted+regenerated controller.capnp.{c,h}) both by implementer and reviewer. Full project build clean (controller/apiserver-CGO/scheduler/worker/xceadmin), 1075/1075 controller_unit_test passing, review APPROVED first pass with reviewer going beyond implementer's own verification (real capn_resolve()-omission and placeholder-removal breakage reproduced firsthand). Zero existing structs/fields touched._ - _Depends: 6_ - [ ] 7.2 Define new event payload types diff --git a/src/controller/testsuite/test_e2e_capnp.c b/src/controller/testsuite/test_e2e_capnp.c index 1d98eadf..731d048f 100644 --- a/src/controller/testsuite/test_e2e_capnp.c +++ b/src/controller/testsuite/test_e2e_capnp.c @@ -331,6 +331,220 @@ static void test_data_job_array_variant_round_trip(void) printf("test_data_job_array_variant_round_trip: PASS\n"); } +/* ------------------------------------------------------------------ */ +/* Test 7: ReplayHeader + SnapshotEntry (replay stream envelope schema, */ +/* scheduler-core task 7.1, Req 5 AC 20) round-trip through the wire. */ +/* */ +/* Built entirely through the low-level capnp accessor API (new_X / */ +/* X_get_* / X_set_*, all generated by $C.fieldgetset) instead of the */ +/* encode_X/decode_X "codec" bridge to a mapname'd C struct. The codec */ +/* bridge is a documented no-op for Data fields in this codebase's */ +/* capnpc-c setup (see decode_UpgradeConfigRequest's empty body above, */ +/* and request.c's xce_replconfigpush_req_raw comment) -- encoding */ +/* ReplayHeader.configHash (a Data field) through it would silently */ +/* write whatever garbage sits in an uninitialized native `struct */ +/* ReplayHeader` local. Going through the accessors avoids that pitfall */ +/* entirely and, since no "real" application C struct exists yet for */ +/* these frame-only replay-envelope types (task 9+ builds the actual */ +/* reader/writer), is the correct level to prove the SCHEMA/wire format */ +/* itself round-trips -- which is all task 7.1 owns. */ +/* ------------------------------------------------------------------ */ + +static void test_replay_header_snapshot_entry_round_trip(void) +{ + /* Arrange + Act (message 1: ReplayHeader) */ + struct capn c1; + capn_init_malloc(&c1); + struct capn_segment *cs1 = capn_root(&c1).seg; + + ReplayCluster_ptr cluster = new_ReplayCluster(cs1); + capn_text cname = { 7, "xce-lab", NULL }; + ReplayCluster_set_name(cluster, cname); + capn_ptr ctrl_list = capn_new_ptr_list(cs1, 2); + capn_setp(ctrl_list, 0, capn_new_string(cs1, "host1:9991", 10)); + capn_setp(ctrl_list, 1, capn_new_string(cs1, "host2:9992", 10)); + ReplayCluster_set_controllers(cluster, ctrl_list); + capn_text rhost = { 5, "host1", NULL }; + ReplayCluster_set_recordingHost(cluster, rhost); + + uint8_t hash[32]; + for (int i = 0; i < 32; i++) + hash[i] = (uint8_t)(i * 3 + 1); + capn_list8 hash_list = capn_new_list8(cs1, 32); + capn_setv8(hash_list, 0, hash, 32); + capn_data hash_data; + hash_data.p = hash_list.p; + + ReplayHeader_ptr hdr = new_ReplayHeader(cs1); + capn_text rv = { 5, "1.0.0", NULL }; + ReplayHeader_set_recorderVersion(hdr, rv); + ReplayHeader_set_configHash(hdr, hash_data); + ReplayHeader_set_cluster(hdr, cluster); + ReplayHeader_set_recordingStartTs(hdr, 123456789ULL); + capn_text wall = { 20, "2026-08-04T00:00:00Z", NULL }; + ReplayHeader_set_recordingStartWall(hdr, wall); + capn_text policy = { 12, "builtin_fcfs", NULL }; + ReplayHeader_set_schedPolicy(hdr, policy); + + capn_setp(capn_root(&c1), 0, hdr.p); + int64_t sz1 = capn_size(&c1); + TEST_ASSERT_TRUE(sz1 > 0); + uint8_t *buf1 = (uint8_t *)malloc((size_t)sz1); + TEST_ASSERT_NOT_NULL(buf1); + capn_write_mem(&c1, buf1, (size_t)sz1, 0); + capn_free(&c1); + + /* Arrange + Act (message 2: SnapshotEntry -- MUST be the first data + * message after ReplayHeader per Req 5 AC 20's stream framing rule; + * encoded here as an independent message buffer since the file-level + * multi-message concatenation is task 9+ reader/writer logic, not + * this schema-only task's concern). */ + struct capn c2; + capn_init_malloc(&c2); + struct capn_segment *cs2 = capn_root(&c2).seg; + + SnapshotData_ptr sdata = new_SnapshotData(cs2); + SnapshotEntry_ptr snap = new_SnapshotEntry(cs2); + SnapshotEntry_set_ts(snap, 99999ULL); + SnapshotEntry_set_data(snap, sdata); + + capn_setp(capn_root(&c2), 0, snap.p); + int64_t sz2 = capn_size(&c2); + TEST_ASSERT_TRUE(sz2 > 0); + uint8_t *buf2 = (uint8_t *)malloc((size_t)sz2); + TEST_ASSERT_NOT_NULL(buf2); + capn_write_mem(&c2, buf2, (size_t)sz2, 0); + capn_free(&c2); + + /* Act: decode both messages back */ + struct capn d1; + TEST_ASSERT_EQUAL_INT(0, capn_init_mem(&d1, buf1, (size_t)sz1, 0)); + ReplayHeader_ptr rhdr; + rhdr.p = capn_getp(capn_root(&d1), 0, 1); + + struct capn d2; + TEST_ASSERT_EQUAL_INT(0, capn_init_mem(&d2, buf2, (size_t)sz2, 0)); + SnapshotEntry_ptr rsnap; + rsnap.p = capn_getp(capn_root(&d2), 0, 1); + + /* Assert: ReplayHeader fields survive field-for-field */ + capn_text out_rv = ReplayHeader_get_recorderVersion(rhdr); + TEST_ASSERT_EQUAL_INT(5, out_rv.len); + TEST_ASSERT_EQUAL_STRING("1.0.0", out_rv.str); + + capn_data out_hash = ReplayHeader_get_configHash(rhdr); + TEST_ASSERT_EQUAL_INT(32, capn_len((capn_list8){ out_hash.p })); + uint8_t out_hash_bytes[32]; + TEST_ASSERT_EQUAL_INT(32, capn_getv8((capn_list8){ out_hash.p }, 0, + out_hash_bytes, 32)); + TEST_ASSERT_EQUAL_MEMORY(hash, out_hash_bytes, 32); + + TEST_ASSERT_EQUAL_UINT64(123456789ULL, + ReplayHeader_get_recordingStartTs(rhdr)); + capn_text out_wall = ReplayHeader_get_recordingStartWall(rhdr); + TEST_ASSERT_EQUAL_STRING("2026-08-04T00:00:00Z", out_wall.str); + capn_text out_policy = ReplayHeader_get_schedPolicy(rhdr); + TEST_ASSERT_EQUAL_STRING("builtin_fcfs", out_policy.str); + + ReplayCluster_ptr out_cluster = ReplayHeader_get_cluster(rhdr); + /* A nested-struct field getter (capn_getp(..., resolve=0)) can return an + * unresolved far pointer; per-field accessors that index raw struct + * offsets (capn_read*) do NOT auto-resolve it (unlike capn_get_text/ + * capn_get_data, which resolve internally), so any nested struct + * pointer must be explicitly resolved before its own field accessors + * are used. */ + capn_resolve(&out_cluster.p); + capn_text out_cname = ReplayCluster_get_name(out_cluster); + TEST_ASSERT_EQUAL_STRING("xce-lab", out_cname.str); + capn_ptr out_ctrl_list = ReplayCluster_get_controllers(out_cluster); + capn_resolve(&out_ctrl_list); + TEST_ASSERT_EQUAL_INT(2, out_ctrl_list.len); + capn_text text_default = { 0, "", NULL }; + capn_text c0 = capn_get_text(out_ctrl_list, 0, text_default); + capn_text c1t = capn_get_text(out_ctrl_list, 1, text_default); + TEST_ASSERT_EQUAL_STRING("host1:9991", c0.str); + TEST_ASSERT_EQUAL_STRING("host2:9992", c1t.str); + capn_text out_rhost = ReplayCluster_get_recordingHost(out_cluster); + TEST_ASSERT_EQUAL_STRING("host1", out_rhost.str); + + /* Assert: SnapshotEntry -- the required Observable of task 7.1 + * (Req 5 AC 20): a minimal header-plus-snapshot-entry byte stream + * round-trips through encode/decode. */ + TEST_ASSERT_EQUAL_UINT64(99999ULL, SnapshotEntry_get_ts(rsnap)); + SnapshotData_ptr out_sdata = SnapshotEntry_get_data(rsnap); + TEST_ASSERT_FALSE(out_sdata.p.type == CAPN_NULL); + + capn_free(&d1); + capn_free(&d2); + free(buf1); + free(buf2); + printf("test_replay_header_snapshot_entry_round_trip: PASS\n"); +} + +/* ------------------------------------------------------------------ */ +/* Test 8: SessionSummary (fully-real, all-scalar struct) round-trips. */ +/* Bonus coverage beyond task 7.1's required Observable -- proves the */ +/* rest of the new schema surface (ReplaySession/ReplayDecision/ */ +/* ReplayEvent) actually compiles and links, not just SnapshotEntry. */ +/* ------------------------------------------------------------------ */ + +static void test_session_summary_round_trip(void) +{ + struct capn c; + capn_init_malloc(&c); + struct capn_segment *cs = capn_root(&c).seg; + + SessionSummary_ptr summary = new_SessionSummary(cs); + SessionSummary_set_durationNs(summary, 5000000000ULL); + SessionSummary_set_nDecisions(summary, 42u); + SessionSummary_set_nDispatched(summary, 40u); + SessionSummary_set_nPended(summary, 2u); + + ReplayDecision_list decisions = new_ReplayDecision_list(cs, 1); + set_ReplayDecision(&(struct ReplayDecision){ .ts = 111ULL }, decisions, 0); + + ReplaySession_ptr sess = new_ReplaySession(cs); + ReplaySession_set_beginRound(sess, 7ULL); + ReplaySession_set_decisions(sess, decisions); + ReplaySession_set_summary(sess, summary); + + capn_setp(capn_root(&c), 0, sess.p); + int64_t sz = capn_size(&c); + TEST_ASSERT_TRUE(sz > 0); + uint8_t *buf = (uint8_t *)malloc((size_t)sz); + TEST_ASSERT_NOT_NULL(buf); + capn_write_mem(&c, buf, (size_t)sz, 0); + capn_free(&c); + + struct capn d; + TEST_ASSERT_EQUAL_INT(0, capn_init_mem(&d, buf, (size_t)sz, 0)); + ReplaySession_ptr rsess; + rsess.p = capn_getp(capn_root(&d), 0, 1); + + TEST_ASSERT_EQUAL_UINT64(7ULL, ReplaySession_get_beginRound(rsess)); + SessionSummary_ptr out_summary = ReplaySession_get_summary(rsess); + /* See the matching comment in the ReplayHeader test above: a nested + * struct pointer returned from a generated X_get_Y() accessor may be an + * unresolved far pointer and must be resolved before its own per-field + * accessors (capn_read*-based) are used. */ + capn_resolve(&out_summary.p); + TEST_ASSERT_EQUAL_UINT64(5000000000ULL, + SessionSummary_get_durationNs(out_summary)); + TEST_ASSERT_EQUAL_UINT32(42u, SessionSummary_get_nDecisions(out_summary)); + TEST_ASSERT_EQUAL_UINT32(40u, SessionSummary_get_nDispatched(out_summary)); + TEST_ASSERT_EQUAL_UINT32(2u, SessionSummary_get_nPended(out_summary)); + + ReplayDecision_list out_decisions = ReplaySession_get_decisions(rsess); + TEST_ASSERT_EQUAL_INT(1, capn_len(out_decisions)); + struct ReplayDecision out_dec; + get_ReplayDecision(&out_dec, out_decisions, 0); + TEST_ASSERT_EQUAL_UINT64(111ULL, out_dec.ts); + + capn_free(&d); + free(buf); + printf("test_session_summary_round_trip: PASS\n"); +} + /* ------------------------------------------------------------------ */ /* Suite entry point */ /* ------------------------------------------------------------------ */ @@ -346,4 +560,6 @@ void test_e2e_capnp(void) RUN_TEST(test_request_vote_last_log_tuple_round_trip); RUN_TEST(test_wal_catchup_append_term_fields_round_trip); RUN_TEST(test_data_job_array_variant_round_trip); + RUN_TEST(test_replay_header_snapshot_entry_round_trip); + RUN_TEST(test_session_summary_round_trip); } diff --git a/src/include/typemap.h b/src/include/typemap.h index 42bc3947..bbb301a7 100644 --- a/src/include/typemap.h +++ b/src/include/typemap.h @@ -541,4 +541,88 @@ struct ctrl_hello_reply_s { uint64_t config_tag; /* primary's current config tag */ }; +/* ── Replay stream envelope frames (scheduler-core Group 7 task 7.1, + * Req 5 AC 20) ────────────────────────────────────────────────────────── + * Frame-only wire types for the `.replay` stream envelope -- same "no live + * C struct" convention as capnp_snap_hdr_t/capnp_persist_cfg_t/ + * capnp_host_admin_state_t above (see controller.capnp's comment at + * `struct ReplayCluster` for the full Group 7 task split). The C-side + * reader/writer that actually populates/consumes these (recording, + * `.replay` file I/O) is task 9+, out of this schema-only task's scope. */ +typedef struct { + char *name; + char **controllers; + int n_controllers; + char *recording_host; +} replay_cluster_t; + +typedef struct { + char *recorder_version; + /* configHash is a Data field -- the generated encode_/decode_ codec is + * a documented no-op for Data fields in this codebase's capnpc-c setup + * (same limitation as e.g. upgrade_config_req_s.sha256/bytes above); + * real byte extraction is task 9+ hand-written code, same raw_sha256- + * style pattern used throughout this file. This pointer/len pair exists + * only so the mapname target type is well-formed; nothing here reads it + * yet. */ + uint8_t *config_hash; + size_t config_hash_len; + replay_cluster_t *cluster; + uint64_t recording_start_ts; + char *recording_start_wall; + char *sched_policy; +} replay_header_t; + +/* Placeholder -- task 7.3 replaces this wholesale (not merely appends to) + * with the real 11-field snapshot struct (config/jobs/queues/workers/ + * resources/resvals/users/projects/prngSeed/internalState/internalVersion, + * design.md's Wire Protocol > ReplayFormat section). The single reserved + * byte below has no capnp counterpart -- SnapshotData has zero capnp fields + * in this task, so the generated encode_SnapshotData/decode_SnapshotData + * bodies are no-ops regardless of what this C struct contains. */ +typedef struct { + uint8_t _reserved; +} sched_snapshot_data_t; + +typedef struct { + uint64_t ts; + sched_snapshot_data_t *data; +} replay_snapshot_entry_t; + +typedef struct { + uint64_t ts; + int body_which; + union { + xce_conf4sched_t *config_push; + xce_jobs_data_t *jobs_push; + xce_workers_data_t *workers_push; + xce_resource_data_t *resources_push; + xce_resval_list_t *resvals_full; + xce_resval_list_t *resvals_delta; + xce_queue_t *queue_update; + xce_job_rusages_t *job_rusage; + /* endSession: Void, no storage */ + } body; +} replay_event_t; + +typedef struct { + uint64_t duration_ns; + uint32_t n_decisions; + uint32_t n_dispatched; + uint32_t n_pended; +} session_summary_t; + +/* Placeholder -- task 7.3 adds the `body` union (dispatch/pend/preempt) + * here once DispatchDecision/PendDecision/PreemptDecision exist. */ +typedef struct { + uint64_t ts; +} replay_decision_t; + +typedef struct { + uint64_t begin_round; + replay_decision_t **decisions; + int n_decisions; + session_summary_t *summary; +} replay_session_t; + #endif diff --git a/src/schema/controller.capnp b/src/schema/controller.capnp index d65f1025..5eb5f20e 100644 --- a/src/schema/controller.capnp +++ b/src/schema/controller.capnp @@ -977,3 +977,108 @@ struct CtrlHelloReply $C.mapname("ctrl_hello_reply_t") { needsSnapshot @1 :Bool $C.mapname("needs_snapshot"); configTag @2 :UInt64 $C.mapname("config_tag"); } + +# ── Replay stream envelope schema (scheduler-core Group 7 task 7.1, +# Req 5 AC 20) ──────────────────────────────────────────────────────────── +# `.replay` file format: a single stream of consecutive capnp messages -- +# ReplayHeader (msg 1) -> SnapshotEntry (msg 2, MUST be the first data +# message) -> ReplayEvent/ReplaySession cycles -> periodic SnapshotEntry +# checkpoints -> final ReplayEvent{endSession}. See requirements.md Req 5 +# AC 20 for the full framing prose; this schema captures only the message +# *shapes* -- the file-level reader/writer that produces/consumes the +# stream is task 9+, entirely out of Group 7's (schema-only) scope. +# +# Struct-ownership split across Group 7 tasks, so a fresh implementer on +# 7.2/7.3/7.4 knows exactly what to add and where. Every later addition is a +# NEW ordinal on a struct THIS task already created -- nothing below is ever +# renumbered or restructured by a later task: +# - task 7.1 (this task): ReplayCluster, ReplayHeader, SnapshotData +# (zero-field placeholder), SnapshotEntry, ReplayEvent (9 members, all +# reusing pre-existing structs), SessionSummary (fully real), +# ReplayDecision (ts-only stub), ReplaySession. +# - task 7.2: adds the new JobStateChangeEvent + JobUpdateEvent structs, +# then adds two new union members (jobStateChange, jobUpdate) to +# ReplayEvent's `body` union at new ordinals (10, 11) -- purely additive. +# - task 7.3: adds SnapshotData's real 11 fields (Req 5 AC 21) starting at +# ordinal @0 -- replacing this task's zero-field placeholder wholesale, +# both in this schema and in typemap.h's sched_snapshot_data_t (which +# has no capnp-visible fields yet, so nothing there is "renumbered" +# either). task 7.3 also adds DispatchDecision/PendDecision/ +# PreemptVictim/PreemptDecision, then adds a `body :union { dispatch @1: +# DispatchDecision; pend @2: PendDecision; preempt @3: PreemptDecision; +# }` to ReplayDecision -- purely additive new ordinals on the ts@0 +# struct this task creates (see the JobControl/JoinData `:union` group +# precedent above: a union group name never consumes its own ordinal, +# only its member variants do, so this is safe to bolt on later). + +struct ReplayCluster $C.mapname("replay_cluster_t") { + name @0: Text; + controllers @1: List(Text) $C.maplistcount("n_controllers"); # ["host1:9991", "host2:9992", "host3:9993"] + recordingHost @2: Text $C.mapname("recording_host"); # controller hostname that made the recording +} + +struct ReplayHeader $C.mapname("replay_header_t") { + recorderVersion @0: Text $C.mapname("recorder_version"); # scheduler binary version that wrote this file + configHash @1: Data $C.mapname("config_hash"); # sha256 of xce_conf4sched_t + cluster @2: ReplayCluster; + recordingStartTs @3: UInt64 $C.mapname("recording_start_ts"); # logical time recording started (ns) + recordingStartWall @4: Text $C.mapname("recording_start_wall"); # wall-clock time recording started (ISO 8601) + schedPolicy @5: Text $C.mapname("sched_policy"); # active picker plugin name (e.g. "builtin_fcfs") +} + +# Placeholder -- task 7.3 replaces this with the full snapshot schema +# (Req 5 AC 21): config, jobs, queues, workers, resources, resvals, users, +# projects, prngSeed, internalState, internalVersion (see design.md's Wire +# Protocol > ReplayFormat section for the already-committed 11-field +# layout). Deliberately left with ZERO capnp fields here (instead of a +# placeholder scalar) so task 7.3 is free to start numbering its real +# fields at ordinal @0 exactly as design.md specifies, with no renumbering. +struct SnapshotData $C.mapname("sched_snapshot_data_t") { +} + +struct SnapshotEntry $C.mapname("replay_snapshot_entry_t") { + ts @0: UInt64; # logical time of snapshot + data @1: SnapshotData; # full core state +} + +# ReplayEvent.body reuses 7 pre-existing wire structs verbatim (Conf4Sched, +# SchedJobData, WorkerData, ResourceData, ResValList x2, QueueConf, +# SchedJobRusages -- verified present in controller.capnp, research.md) plus +# a Void end-of-recording marker. jobStateChange/jobUpdate (task 7.2) are +# added later as new union members at ordinals 10/11, once +# JobStateChangeEvent/JobUpdateEvent exist. +struct ReplayEvent $C.mapname("replay_event_t") { + ts @0: UInt64; # logical time + body :union $C.mapuniontag("body_which") { + configPush @1: Conf4Sched $C.mapname("config_push"); + jobsPush @2: SchedJobData $C.mapname("jobs_push"); + workersPush @3: WorkerData $C.mapname("workers_push"); + resourcesPush @4: ResourceData $C.mapname("resources_push"); + resvalsFull @5: ResValList $C.mapname("resvals_full"); + resvalsDelta @6: ResValList $C.mapname("resvals_delta"); + queueUpdate @7: QueueConf $C.mapname("queue_update"); + jobRusage @8: SchedJobRusages $C.mapname("job_rusage"); + endSession @9: Void; # recording stop marker + } +} + +struct SessionSummary $C.mapname("session_summary_t") { + durationNs @0: UInt64 $C.mapname("duration_ns"); + nDecisions @1: UInt32 $C.mapname("n_decisions"); + nDispatched @2: UInt32 $C.mapname("n_dispatched"); + nPended @3: UInt32 $C.mapname("n_pended"); +} + +# Placeholder -- task 7.3 adds `body :union { dispatch @1: DispatchDecision; +# pend @2: PendDecision; preempt @3: PreemptDecision; }` here once those +# payload types exist (purely additive new ordinals 1-3; ts@0 untouched -- +# see the JobControl/JoinData precedent noted above). +struct ReplayDecision $C.mapname("replay_decision_t") { + ts @0: UInt64; +} + +struct ReplaySession $C.mapname("replay_session_t") { + beginRound @0: UInt64 $C.mapname("begin_round"); # logical time (ts value) + decisions @1: List(ReplayDecision) $C.maplistcount("n_decisions"); + summary @2: SessionSummary; # optional diagnostic metadata +}