feat(scheduler-core): build xceadmin scheduler dump command (task 10.4)

Adds dump() to scheduler.py -- a pure local .replay file decoder, zero
network I/O. Key finding: pycapnp's from_bytes() does not validate a
message against the schema type you ask it to decode as (trial-decoding
SnapshotEntry bytes as ReplayEvent raises nothing, returns plausible
garbage), ruling out trial-decode-and-catch for message-type
discrimination. Instead reads each message's raw segment-table/
root-pointer framing to extract its (dataWordCount, pointerCount) shape,
computed live from the schema, and matches against the four replay
message types' distinct shapes before decoding. AC 20's stream-ordering
rules (msg 1 = ReplayHeader, msg 2 = SnapshotEntry) are enforced
structurally, not inferred.

Adversarial review reproduced a real (if forward-looking) gap: an actual
modified schema with a shape collision caused an uncaught raw capnp
exception to escape dump(). Closed with a pairwise-distinctness
assertion right after the four shapes are computed, raising a typed
ReplayFileError before any message is decoded if a future schema change
ever makes two shapes collide -- restoring dump()'s own docstring
guarantee unconditionally. New regression test simulates the exact
collision the review found.

201 passed, 3 skipped (up from 188 pre-task).
This commit is contained in:
Rongsong Shen 2026-08-06 05:29:15 +00:00
parent b64875f4c3
commit 8034fb6e1f
Signed by: rshen
SSH key fingerprint: SHA256:8Vf2pZ3h6rJa7EhGK+WaAQ9Mta/apZvteW6g9U/WKKk
4 changed files with 679 additions and 13 deletions

View file

@ -281,11 +281,12 @@
- _Boundary: XceAdminSchedulerCLI — not (P): 10.3/10.4 both add functions to the same new scheduler.py module; run sequentially to avoid concurrent edits to one file_ - _Boundary: XceAdminSchedulerCLI — not (P): 10.3/10.4 both add functions to the same new scheduler.py module; run sequentially to avoid concurrent edits to one file_
- _Verified: 2026-08-05 — new `src/xceadmin/src/xceadmin/scheduler.py` (design.md's exact target file/function names) implements `record_start(controller, filename=None)`/`record_stop(controller)`/`record_status(controller)`, all funneled through one shared `_send_admin_request()`: build `SchedAdminRequest` via the real bundled schema (`request.opcode = "schedAdmin"`, `request.body.init("schedadminmsg")`) → `xceadmin.discovery.follow_redirects` (redirect-to-primary, mirrors `push_config`) → `xceadmin.discovery.decode_reply` → extract `reply.data.schedadminreplymsg`. Deliberate, documented divergence from `push_config()`'s "return failure as data" convention: this task's own Observable text explicitly says "raises a typed error on ... an error reply", so a `SchedAdminReply{ok=false}` raises a new `SchedAdminError` (`xcecommon/errors.py`, subclasses `XceAdminError`, carries `error`/`recording`/`file`/`event_count`/`started_at` — the full reply state, not just the reason string, since e.g. "already recording" still reports meaningful current-file state per requirements.md's Reply field semantics table); connection failures (`ConnectionTimeoutError`/`NoPrimaryError`/`TooManyRedirectsError`) and decode failures (`ReplyDecodeError`) reuse the existing typed errors `follow_redirects`/`decode_reply` already raise, unchanged. On success returns a new frozen `RecordResult` dataclass (`recording`/`file`/`event_count`/`started_at`; `ok`/`error` omitted as trivially-true/empty post-raise). `test_errors.py` extended with `SchedAdminError` in `ALL_SUBCLASSES` plus its own carries-state test (matching the existing exhaustive per-subclass pattern). New `test_scheduler.py` (mirrors `test_config_push_mock.py`'s mock-TLS-server pattern): success for all 3 actions (incl. explicit vs. auto-generated filename), one redirect-following case (`record_start` standby→primary), `ok=false` raising `SchedAdminError` for both `record_start`("already recording") and `record_stop`("not recording") plus a parametrized sweep of the Error codes reference table's other reasons, wrong-reply-variant → `ReplyDecodeError`, and a monkeypatched connection-timeout → `NoPrimaryError` (composition test, same shape as `push_config`'s own). 188 passed / 3 skipped (pre-existing live-cluster E2E tests, unaffected) via `meson compile -C xce.build xceadmin` — the real xceadmin test-running command (custom_target wrapping `meson-build.sh`'s pytest run; confirmed no `meson test --suite xceadmin` exists, `test()` is never used for this Python-only subdir). Noted the bundled schema copy under the live checkout (`src/xceadmin/src/xcecommon/schema_files/controller.capnp`) is an untracked, pre-existing-stale build artifact re-synced from `src/schema/controller.capnp` on every `meson compile` run — not touched, not a bug introduced by this task (see cerebrum.md). Independent adversarial review confirmed every claim above against source and a fresh test run (188/188+3 skipped, exact match), including the raise-vs-return divergence being a defensible, well-justified reading of this task's own binding Observable text (not a misreading of `push_config()`'s convention) — APPROVED, no remediation required. One non-blocking coverage gap noted for a future pass: `_send_admin_request()`'s `reply.status != 0` defensive branch has no dedicated test, unlike `push_config()`'s analogous coverage. This entry was originally self-authored by the implementer rather than the orchestrator (a process deviation, flagged and reviewed accordingly) — content re-verified accurate and left in place rather than rewritten._ - _Verified: 2026-08-05 — new `src/xceadmin/src/xceadmin/scheduler.py` (design.md's exact target file/function names) implements `record_start(controller, filename=None)`/`record_stop(controller)`/`record_status(controller)`, all funneled through one shared `_send_admin_request()`: build `SchedAdminRequest` via the real bundled schema (`request.opcode = "schedAdmin"`, `request.body.init("schedadminmsg")`) → `xceadmin.discovery.follow_redirects` (redirect-to-primary, mirrors `push_config`) → `xceadmin.discovery.decode_reply` → extract `reply.data.schedadminreplymsg`. Deliberate, documented divergence from `push_config()`'s "return failure as data" convention: this task's own Observable text explicitly says "raises a typed error on ... an error reply", so a `SchedAdminReply{ok=false}` raises a new `SchedAdminError` (`xcecommon/errors.py`, subclasses `XceAdminError`, carries `error`/`recording`/`file`/`event_count`/`started_at` — the full reply state, not just the reason string, since e.g. "already recording" still reports meaningful current-file state per requirements.md's Reply field semantics table); connection failures (`ConnectionTimeoutError`/`NoPrimaryError`/`TooManyRedirectsError`) and decode failures (`ReplyDecodeError`) reuse the existing typed errors `follow_redirects`/`decode_reply` already raise, unchanged. On success returns a new frozen `RecordResult` dataclass (`recording`/`file`/`event_count`/`started_at`; `ok`/`error` omitted as trivially-true/empty post-raise). `test_errors.py` extended with `SchedAdminError` in `ALL_SUBCLASSES` plus its own carries-state test (matching the existing exhaustive per-subclass pattern). New `test_scheduler.py` (mirrors `test_config_push_mock.py`'s mock-TLS-server pattern): success for all 3 actions (incl. explicit vs. auto-generated filename), one redirect-following case (`record_start` standby→primary), `ok=false` raising `SchedAdminError` for both `record_start`("already recording") and `record_stop`("not recording") plus a parametrized sweep of the Error codes reference table's other reasons, wrong-reply-variant → `ReplyDecodeError`, and a monkeypatched connection-timeout → `NoPrimaryError` (composition test, same shape as `push_config`'s own). 188 passed / 3 skipped (pre-existing live-cluster E2E tests, unaffected) via `meson compile -C xce.build xceadmin` — the real xceadmin test-running command (custom_target wrapping `meson-build.sh`'s pytest run; confirmed no `meson test --suite xceadmin` exists, `test()` is never used for this Python-only subdir). Noted the bundled schema copy under the live checkout (`src/xceadmin/src/xcecommon/schema_files/controller.capnp`) is an untracked, pre-existing-stale build artifact re-synced from `src/schema/controller.capnp` on every `meson compile` run — not touched, not a bug introduced by this task (see cerebrum.md). Independent adversarial review confirmed every claim above against source and a fresh test run (188/188+3 skipped, exact match), including the raise-vs-return divergence being a defensible, well-justified reading of this task's own binding Observable text (not a misreading of `push_config()`'s convention) — APPROVED, no remediation required. One non-blocking coverage gap noted for a future pass: `_send_admin_request()`'s `reply.status != 0` defensive branch has no dedicated test, unlike `push_config()`'s analogous coverage. This entry was originally self-authored by the implementer rather than the orchestrator (a process deviation, flagged and reviewed accordingly) — content re-verified accurate and left in place rather than rewritten._
- [ ] 10.4 Build the xceadmin scheduler dump command - [x] 10.4 Build the xceadmin scheduler dump command
- Implement a local-file decoder that reads a replay file's message stream and formats it to human-readable text — no network connection involved - Implement a local-file decoder that reads a replay file's message stream and formats it to human-readable text — no network connection involved
- Observable: running dump against a fixture replay file produces text output listing the header fields, snapshot entry, and each event/session/decision in the file - Observable: running dump against a fixture replay file produces text output listing the header fields, snapshot entry, and each event/session/decision in the file
- _Requirements: 5.20_ - _Requirements: 5.20_
- _Depends: 7.5_ - _Depends: 7.5_
- _Verified: 2026-08-05 — `dump(replay_file_path)` (extends `scheduler.py`, task 10.3's file) is a pure local file decoder, zero network I/O (confirmed no `wire`/`discovery`/`connect` reachable from its code path). Key finding, empirically reproduced: pycapnp's `from_bytes()` does NOT validate a message's bytes against the schema you ask it to decode as — decoding real `SnapshotEntry` bytes as `ReplayEvent` raises nothing and returns plausible-looking garbage field values, ruling out trial-decode-and-catch as a message-type-discrimination strategy. Instead, `dump()` reads each message's raw segment-table/root-pointer framing directly (`_peek_message()`) to extract its `(dataWordCount, pointerCount)` shape — computed live from each candidate schema's `cls.schema.node.struct`, never hardcoded — and matches against the four replay message types' actual shapes (`ReplayHeader=(1,5)`, `SnapshotEntry=(1,1)`, `ReplayEvent=(2,1)`, `ReplaySession=(1,2)`) before ever calling the real decoder; AC 20's stream-ordering rules (message 1 = ReplayHeader, message 2 = SnapshotEntry) are enforced via position-restricted shape dicts, not inferred. Adversarial review independently reproduced the trial-decode-unsafe finding from scratch (own probe script against the live schema) and went further: built an actual modified schema (one extra field on SnapshotEntry) and reproduced a real shape collision end-to-end, finding it caused an uncaught raw capnp exception to escape `dump()` — a genuine, if forward-looking, gap since today's four shapes are independently confirmed pairwise-distinct. Closed immediately: added a pairwise-distinctness assertion right after the four shapes are computed, raising a typed `ReplayFileError(reason="ambiguous message shapes")` naming the colliding types before any message is decoded, restoring `dump()`'s own docstring guarantee ("never lets a raw exception escape") unconditionally rather than only-if-the-schema-stays-lucky; added `test_dump_raises_on_ambiguous_message_shapes` (monkeypatches `_shape_of()` to simulate the exact collision the review reproduced) proving the guard fires — independently re-verified by me via a fresh `meson compile -C xce.build xceadmin` run (201 passed, 3 skipped, up from 188 pre-task) before marking complete. Test fixture (`_build_full_replay_stream()`, `test_scheduler.py`) is hand-built via pycapnp's write-side API since no `.replay` fixture existed anywhere in the repo yet (no C-side simulation driver built so far); confirmed byte-for-byte framing-compatible with `recording.c`'s real `write_capnp_message()` (plain concatenated `to_bytes()` output, zero extra framing). Snapshot entries are summarized (counts/names) rather than exhaustively enumerated, decisions are fully enumerated per-item — a defensible reading of AC 20's "listing... the snapshot entry" (singular) vs. "each event/session/decision" (per-item) wording, confirmed reasonable by review, not blocking._
- _Boundary: XceAdminSchedulerCLI — not (P), same file as 10.3 (see 10.3)_ - _Boundary: XceAdminSchedulerCLI — not (P), same file as 10.3 (see 10.3)_
- [ ] 10.5 Wire the xceadmin CLI commands - [ ] 10.5 Wire the xceadmin CLI commands

View file

@ -3,10 +3,14 @@ Requirements 5.10, 5.20).
This module is the shared home for the scheduler recording-control client This module is the shared home for the scheduler recording-control client
functions (``record_start``/``record_stop``/``record_status``, task 10.3) functions (``record_start``/``record_stop``/``record_status``, task 10.3)
AND the local ``.replay`` file decoder (``dump()``, task 10.4, NOT built AND the local ``.replay`` file decoder (``dump()``, task 10.4) --
here) -- design.md's File Structure Plan places both in this one file, design.md's File Structure Plan places both in this one file, mirroring
mirroring ``push.py``'s split (all business logic here, ``cli.py`` stays a ``push.py``'s split (all business logic here, ``cli.py`` stays a thin
thin ``click`` layer -- task 10.5, also not built here). ``click`` layer -- task 10.5, not built here). ``dump()`` and its helpers
are a completely separate code path from the request/reply functions
above: no network, no ``xcecommon.wire``, no ``xceadmin.discovery`` --
see the "dump(): local-file .replay decoder" section below for its own
docstring and design rationale.
``record_start``/``record_stop``/``record_status`` mirror ``push.py``'s ``record_start``/``record_stop``/``record_status`` mirror ``push.py``'s
``push_config()`` pipeline exactly: build request bytes with the real ``push_config()`` pipeline exactly: build request bytes with the real
@ -58,10 +62,11 @@ wire struct verbatim).
from __future__ import annotations from __future__ import annotations
import struct
from dataclasses import dataclass from dataclasses import dataclass
from xcecommon import wire from xcecommon import wire
from xcecommon.errors import ReplyDecodeError, SchedAdminError from xcecommon.errors import ReplayFileError, ReplyDecodeError, SchedAdminError
from xcecommon.schema import load_schema from xcecommon.schema import load_schema
from xceadmin.discovery import decode_reply, follow_redirects from xceadmin.discovery import decode_reply, follow_redirects
@ -191,3 +196,355 @@ def record_status(controller: str) -> RecordResult:
action" table's own framing of all three actions as one contract. action" table's own framing of all three actions as one contract.
""" """
return _send_admin_request(controller, "recordStatus") return _send_admin_request(controller, "recordStatus")
# ---------------------------------------------------------------------------
# dump(): local-file .replay decoder (task 10.4, Requirement 5.20).
#
# No network involved at all -- this is a completely different code path
# from record_start()/record_stop()/record_status() above, sharing only
# ``xcecommon.schema.load_schema()`` (the same bundled controller.capnp
# every other xceadmin command decodes against).
#
# **The load-bearing technical finding this section is built on** (research
# probe against the real pycapnp/capnproto install in this repo's nix dev
# shell, 2026-08-05 -- see scratchpad probe_capnp.py, not committed):
# requirements.md AC 20's ``.replay`` stream has NO self-describing type tag
# for any message. Position only pins down the first two messages
# (``ReplayHeader`` then ``SnapshotEntry``, AC 20's "MUST be first after
# header"); every message after that is SOME mix of ``ReplayEvent``,
# ``ReplaySession``, and periodic checkpoint ``SnapshotEntry`` (AC 11) with
# no further positional guarantee -- confirmed by reading the C-side writer
# (recording.c/recording_events.c/recording_session.c): every message is
# written via ``write_capnp_message()``, which appends ONLY the raw capnp
# message (segment table + data), no discriminator byte of any kind.
#
# The obvious fallback -- "try decoding each message as each candidate type,
# see which one parses" -- is NOT reliable and was empirically ruled out,
# not just assumed unreliable: pycapnp's dynamic API is deliberately lenient
# about schema mismatches (this is how capnp's forward/backward schema
# compatibility works). Concretely verified: a real ``SnapshotEntry``'s
# bytes decoded AS ``ReplayEvent`` raise no exception at all and produce a
# perfectly plausible-looking ``body.which() == 'configPush'``; the reverse
# (``ReplayEvent`` bytes decoded as ``ReplaySession``) likewise raises
# nothing and returns a plausible (wrong) ``beginRound``/empty
# ``decisions``. ``.total_size`` is IDENTICAL across every candidate schema
# for the same bytes (it reflects the underlying serialized data, not the
# schema used to read it) -- also not a usable signal. Only one exception
# was ever observed, and only for a specific field-type collision (a List
# pointer landing where Text was expected) -- not something to build a
# general discriminator on.
#
# What DOES reliably work, verified directly: each of the four top-level
# ``.replay`` message types has a DISTINCT (dataWordCount, pointerCount)
# struct shape -- ``ReplayHeader``=(1,5), ``SnapshotEntry``=(1,1),
# ``ReplayEvent``=(2,1), ``ReplaySession``=(1,2) -- and that shape is
# encoded directly in the message's own root struct pointer (the first word
# of its first segment), independent of whatever schema is later used to
# interpret the fields. ``_peek_message()`` below reads ONLY that framing
# metadata (the same segment-table framing pycapnp's own
# ``read_multiple``/``read_multiple_bytes`` already parses internally to
# find message boundaries -- this is not a hand-rolled *content* decoder,
# just the standard Cap'n Proto stream framing:
# https://capnproto.org/encoding.html#serialization-over-a-stream) to
# determine both a message's total byte length (to advance to the next
# message) AND which of the four known types it actually is, BEFORE ever
# asking pycapnp to interpret its fields. All field-level decoding below
# still goes through pycapnp's real generated schema classes
# (``mod.ReplayHeader.from_bytes()`` etc.) -- ``_peek_message()`` only ever
# decides WHICH class to call that with.
#
# This position-based-plus-shape-check design is very likely what the
# C-side simulation driver (not yet built -- Group 11+, out of this task's
# scope) will also need, since the wire format itself offers no other
# option -- documented here for whoever builds that driver next.
# ---------------------------------------------------------------------------
_STRUCT_POINTER_KIND = 0
def _peek_message(data: bytes, offset: int) -> tuple[int, int, int, int]:
"""Read the Cap'n Proto stream-framing metadata for the message
starting at ``data[offset:]`` -- returns ``(total_byte_length, kind,
data_word_count, pointer_count)`` -- without interpreting any field via
any schema (module section docstring above).
Raises :class:`struct.error` if fewer bytes remain than the segment
table itself claims to need (truncated/garbage input) -- callers wrap
this as :class:`~xcecommon.errors.ReplayFileError`.
"""
(seg_count_minus_one,) = struct.unpack_from("<I", data, offset)
seg_count = seg_count_minus_one + 1
sizes = struct.unpack_from(f"<{seg_count}I", data, offset + 4)
table_words = 1 + seg_count # 4-byte units: the count field + N size fields
table_bytes = ((table_words + 1) // 2) * 8 # padded to an 8-byte boundary
seg0_start = offset + table_bytes
(root_ptr,) = struct.unpack_from("<Q", data, seg0_start)
kind = root_ptr & 0x3
data_words = (root_ptr >> 32) & 0xFFFF
pointer_count = (root_ptr >> 48) & 0xFFFF
total_bytes = table_bytes + sum(sizes) * 8
return total_bytes, kind, data_words, pointer_count
def _shape_of(cls) -> tuple[int, int]:
"""``(dataWordCount, pointerCount)`` for a loaded capnp struct class --
computed from the live schema (never hardcoded), so this stays correct
if the schema evolves.
"""
node = cls.schema.node.struct
return node.dataWordCount, node.pointerCount
def dump(replay_file_path: str) -> str:
"""Decode a local ``.replay`` file (task 10.4's Observable: "running
dump against a fixture replay file produces text output listing the
header fields, snapshot entry, and each event/session/decision in the
file") -- no network connection involved.
Returns the full human-readable text as one string (mirrors this
task's own Observable wording of "text output", not a structured
return type -- CLI wiring/printing is task 10.5's job, not this
function's).
Raises :class:`~xcecommon.errors.ReplayFileError` if the path can't be
opened/read, the file is empty, a message is truncated, or a message's
wire shape doesn't match any known ``.replay`` message type (module
section docstring above) -- never lets a raw exception (``OSError``,
``struct.error``, a raw capnp exception) escape.
"""
try:
with open(replay_file_path, "rb") as f:
data = f.read()
except OSError as exc:
raise ReplayFileError(
f"cannot read replay file {replay_file_path!r}: {exc}",
path=replay_file_path,
reason=str(exc),
) from exc
if not data:
raise ReplayFileError(
f"replay file {replay_file_path!r} is empty", path=replay_file_path, reason="empty file"
)
mod = load_schema()
header_shape = _shape_of(mod.ReplayHeader)
snapshot_shape = _shape_of(mod.SnapshotEntry)
event_shape = _shape_of(mod.ReplayEvent)
session_shape = _shape_of(mod.ReplaySession)
# Message-type discrimination below relies entirely on these four shapes
# staying pairwise distinct (trial-decoding a message as the wrong schema
# type does not raise -- see this module's dump() docstring). A future
# schema edit that grows one struct's field count into another's shape
# would otherwise route a message to the wrong decoder and surface as an
# uncaught, undiagnosable capnp exception deep inside from_bytes(). Fail
# loud and specific here instead, at the one place all four shapes are
# known, before any message is decoded.
named_shapes = {
"ReplayHeader": header_shape,
"SnapshotEntry": snapshot_shape,
"ReplayEvent": event_shape,
"ReplaySession": session_shape,
}
seen_shapes: dict[tuple[int, int], str] = {}
for name, shape in named_shapes.items():
collided_with = seen_shapes.get(shape)
if collided_with is not None:
raise ReplayFileError(
f"cannot dump {replay_file_path!r}: schema ambiguity -- "
f"{name} and {collided_with} both have shape {shape} "
"(dataWords, pointers), so messages of one type cannot be "
"reliably distinguished from the other",
path=replay_file_path,
reason="ambiguous message shapes",
)
seen_shapes[shape] = name
pos = 0
msg_no = 0
def _next_message(expected_label: str, shape_to_cls: dict[tuple[int, int], object]):
nonlocal pos, msg_no
msg_no += 1
try:
total_len, kind, data_words, pointer_count = _peek_message(data, pos)
except struct.error as exc:
raise ReplayFileError(
f"replay file {replay_file_path!r} is truncated or malformed at "
f"message {msg_no} (byte offset {pos}): {exc}",
path=replay_file_path,
reason=str(exc),
) from exc
if total_len <= 0 or pos + total_len > len(data):
raise ReplayFileError(
f"replay file {replay_file_path!r} is truncated: message {msg_no} "
f"declares {total_len} bytes but only {len(data) - pos} remain",
path=replay_file_path,
reason="truncated message",
)
cls = shape_to_cls.get((data_words, pointer_count)) if kind == _STRUCT_POINTER_KIND else None
if cls is None:
raise ReplayFileError(
f"replay file {replay_file_path!r}: message {msg_no} at byte offset "
f"{pos} has an unrecognized shape (kind={kind}, dataWords={data_words}, "
f"pointers={pointer_count}); expected {expected_label}",
path=replay_file_path,
reason="unrecognized message shape",
)
msg_bytes = data[pos : pos + total_len]
pos += total_len
return cls, msg_bytes
lines: list[str] = []
# msg 1 (AC 20): always ReplayHeader.
cls, msg_bytes = _next_message("ReplayHeader", {header_shape: mod.ReplayHeader})
with cls.from_bytes(msg_bytes) as hdr:
lines.extend(_format_header(hdr))
# msg 2 (AC 20): MUST be the initial SnapshotEntry.
cls, msg_bytes = _next_message("SnapshotEntry (initial)", {snapshot_shape: mod.SnapshotEntry})
with cls.from_bytes(msg_bytes) as entry:
lines.append("")
lines.extend(_format_snapshot_entry(entry, "SnapshotEntry (initial)"))
# msg 3+: a mix of ReplayEvent / ReplaySession / periodic checkpoint
# SnapshotEntry (AC 11), in whatever order the recording produced.
trailing_shapes = {
event_shape: mod.ReplayEvent,
session_shape: mod.ReplaySession,
snapshot_shape: mod.SnapshotEntry,
}
while pos < len(data):
cls, msg_bytes = _next_message(
"ReplayEvent, ReplaySession, or SnapshotEntry (checkpoint)", trailing_shapes
)
lines.append("")
if cls is mod.ReplayEvent:
with cls.from_bytes(msg_bytes) as ev:
lines.extend(_format_event(ev))
elif cls is mod.ReplaySession:
with cls.from_bytes(msg_bytes) as sess:
lines.extend(_format_session(sess))
else:
with cls.from_bytes(msg_bytes) as entry:
lines.extend(_format_snapshot_entry(entry, "SnapshotEntry (checkpoint)"))
return "\n".join(lines) + "\n"
def _format_header(hdr) -> list[str]:
cluster = hdr.cluster
return [
"=== ReplayHeader ===",
f"recorderVersion: {hdr.recorderVersion}",
f"configHash: {hdr.configHash.hex()}",
f"recordingStartTs: {hdr.recordingStartTs}",
f"recordingStartWall: {hdr.recordingStartWall}",
f"schedPolicy: {hdr.schedPolicy}",
f"cluster.name: {cluster.name}",
f"cluster.recordingHost: {cluster.recordingHost}",
f"cluster.controllers: {list(cluster.controllers)}",
]
def _format_snapshot_entry(entry, label: str) -> list[str]:
data = entry.data
lines = [f"=== {label} ===", f"ts: {entry.ts}"]
lines.append(f"config.clustername: {data.config.clustername}")
lines.append(f"config.n_resources: {len(data.config.resources)}")
lines.append(f"config.n_queues: {len(data.config.queues)}")
lines.append(f"jobs.n_total: {data.jobs.ntotal}")
lines.append(f"jobs: {[j.jobid for j in data.jobs.jobs]}")
lines.append(f"queues.n_total: {len(data.queues)}")
for q in data.queues:
lines.append(f" queue: name={q.name} priority={q.priority} status={q.status}")
lines.append(f"workers.n_total: {data.workers.ntotal}")
lines.append(f"workers: {[w.name for w in data.workers.workers]}")
lines.append(f"resources.n_total: {data.resources.ntotal}")
lines.append(f"resources: {[r.name for r in data.resources.resources]}")
lines.append(f"resvals: {len(data.resvals.vals)}")
lines.append(f"users: {[u.name for u in data.users]}")
lines.append(f"projects: {[(p.name, p.priority) for p in data.projects]}")
lines.append(f"prngSeed: {data.prngSeed}")
lines.append(f"internalVersion: {data.internalVersion}")
lines.append(f"internalState: {len(data.internalState)} bytes")
return lines
def _format_event(ev) -> list[str]:
which = ev.body.which()
lines = ["=== ReplayEvent ===", f"ts: {ev.ts}", f"type: {which}"]
if which == "configPush":
lines.append(f" clustername: {ev.body.configPush.clustername}")
elif which == "jobsPush":
jd = ev.body.jobsPush
lines.append(f" n_total: {jd.ntotal}")
lines.append(f" jobids: {[j.jobid for j in jd.jobs]}")
elif which == "workersPush":
wd = ev.body.workersPush
lines.append(f" n_total: {wd.ntotal}")
lines.append(f" names: {[w.name for w in wd.workers]}")
elif which == "resourcesPush":
rd = ev.body.resourcesPush
lines.append(f" n_total: {rd.ntotal}")
lines.append(f" names: {[r.name for r in rd.resources]}")
elif which in ("resvalsFull", "resvalsDelta"):
rv = getattr(ev.body, which)
lines.append(f" count: {len(rv.vals)}")
for v in rv.vals:
lines.append(f" {v.name}@{v.host}")
elif which == "queueUpdate":
q = ev.body.queueUpdate
lines.append(f" name: {q.name} priority={q.priority} status={q.status}")
elif which == "jobRusage":
ru = ev.body.jobRusage
lines.append(f" jobid: {ru.jobid} usages={len(ru.usages)}")
elif which == "jobStateChange":
jsc = ev.body.jobStateChange
lines.append(
f" jobid={jsc.jobid} oldState={jsc.oldState} newState={jsc.newState} exitCode={jsc.exitCode}"
)
elif which == "jobUpdate":
ju = ev.body.jobUpdate
lines.append(f" jobid={ju.jobid} changedFields={ju.changedFields:#x}")
elif which == "endSession":
lines.append(" -- end of recording --")
return lines
def _format_decision(d) -> str:
which = d.body.which()
if which == "dispatch":
alloc = d.body.dispatch.alloc
hosts = [ha.host for ha in alloc.allocs]
return f"ts={d.ts} type=dispatch jobid={alloc.jobid} hosts={hosts}"
if which == "pend":
pd = d.body.pend
return f"ts={d.ts} type=pend jobid={pd.jobid} mainReason={pd.reason.mainReason}"
if which == "preempt":
pe = d.body.preempt
victims = [(v.jobid, v.workerid) for v in pe.victims]
return f"ts={d.ts} type=preempt jobid={pe.jobid} victims={victims} mainReason={pe.reason.mainReason}"
return f"ts={d.ts} type={which}"
def _format_session(sess) -> list[str]:
lines = ["=== ReplaySession ===", f"beginRound: {sess.beginRound}", f"decisions: {len(sess.decisions)}"]
for i, d in enumerate(sess.decisions):
lines.append(f" [{i}] {_format_decision(d)}")
s = sess.summary
lines.append(
f"summary: durationNs={s.durationNs} nDecisions={s.nDecisions} "
f"nDispatched={s.nDispatched} nPended={s.nPended}"
)
return lines

View file

@ -129,6 +129,28 @@ class NoPrimaryError(XceAdminError):
self.port = port self.port = port
class ReplayFileError(XceAdminError):
"""A local ``.replay`` file could not be read or decoded (Requirement
5.20; declared here, raised by a later task -- see
``xceadmin.scheduler.dump()``).
Covers every local-file failure mode ``dump()`` refuses to let crash
uncontrolled: the path doesn't exist / isn't readable, the file is
empty, a message is truncated mid-stream, or a message's wire-level
shape doesn't match any known ``.replay`` stream message type (a
format violation of Requirement 5.20's ordering rules -- e.g. the
second message isn't a ``SnapshotEntry``). Distinct from
:class:`ReplyDecodeError`, which is specifically about the
network-reply envelope -- this is a local-file concern with no
network/reply framing involved at all.
"""
def __init__(self, message: str, path: str, reason: str) -> None:
super().__init__(message)
self.path = path
self.reason = reason
class SchedAdminError(XceAdminError): class SchedAdminError(XceAdminError):
"""A ``SchedAdminRequest`` (scheduler recording control) round-tripped """A ``SchedAdminRequest`` (scheduler recording control) round-tripped
successfully but the reply itself reports failure -- successfully but the reply itself reports failure --

View file

@ -1,5 +1,7 @@
"""Mock-controller tests for ``xceadmin.scheduler``'s record-control client """Mock-controller tests for ``xceadmin.scheduler``'s record-control client
functions (design.md, ``xceadmin.scheduler``; Requirements 5.10). functions (design.md, ``xceadmin.scheduler``; Requirements 5.10) AND
golden-file tests for the local ``.replay`` file decoder, ``dump()`` (task
10.4, Requirement 5.20).
Mirrors ``test_config_push_mock.py``'s mock TLS server pattern exactly Mirrors ``test_config_push_mock.py``'s mock TLS server pattern exactly
(same ``mock_controller``/``wrap_reply``/``tls_certs`` fixtures from (same ``mock_controller``/``wrap_reply``/``tls_certs`` fixtures from
@ -10,11 +12,16 @@ test environment), so these are tested against a real local TLS mock
server built from the actual bundled schema, exactly as server built from the actual bundled schema, exactly as
``test_config_push_mock.py`` does for ``push_config``/``probe_config``. ``test_config_push_mock.py`` does for ``push_config``/``probe_config``.
Only ``record_start``/``record_stop``/``record_status`` (task 10.3) are ``dump()`` tests (bottom of this file) are a completely separate concern
covered here. ``dump()`` (task 10.4, not yet built) will get its own -- no network, no mock controller. No pre-existing ``.replay`` fixture
golden-file tests appended to this file later (design.md's File Structure file exists anywhere in this repo (confirmed by a repo-wide search before
Plan places both in one ``test_scheduler.py``) -- not duplicated or writing these tests) and the C-side simulation driver that would consume
anticipated here. one doesn't exist yet either (Group 11+, out of scope) -- so the fixture
stream is built directly with pycapnp's write-side API
(``mod.X.new_message()`` + ``.to_bytes()``, concatenated with NO extra
framing), mirroring exactly what the real C writer
(``recording.c``/``recording_events.c``/``recording_session.c``'s
``write_capnp_message()``) produces on the wire.
""" """
from __future__ import annotations from __future__ import annotations
@ -27,7 +34,7 @@ from pathlib import Path
import pytest import pytest
from xcecommon import wire from xcecommon import wire
from xcecommon.errors import NoPrimaryError, ReplyDecodeError, SchedAdminError from xcecommon.errors import NoPrimaryError, ReplayFileError, ReplyDecodeError, SchedAdminError
from xcecommon.schema import load_schema from xcecommon.schema import load_schema
from xceadmin import scheduler from xceadmin import scheduler
@ -408,3 +415,282 @@ def test_record_explicit_port_parsed():
host, port = scheduler._split_controller("somehost:1234") host, port = scheduler._split_controller("somehost:1234")
assert host == "somehost" assert host == "somehost"
assert port == 1234 assert port == 1234
# ---------------------------------------------------------------------------
# dump(): local .replay file decoder (task 10.4, Requirement 5.20).
#
# No mock controller involved -- these tests build a ``.replay`` byte
# stream directly with pycapnp and write it to a real file on disk, then
# call ``scheduler.dump()`` against that real path (module docstring).
# ---------------------------------------------------------------------------
def _build_full_replay_stream(mod) -> bytes:
"""Builds one complete, well-formed ``.replay`` stream covering every
message kind AC 20 documents, in the order the real C recorder
produces them: ``ReplayHeader`` -> initial ``SnapshotEntry`` ->
``ReplayEvent`` (jobsPush) -> a periodic checkpoint ``SnapshotEntry``
(task 9.3's feature, AC 11) -> ``ReplaySession`` (one dispatch + one
pend decision) -> ``ReplayEvent{endSession}``. Used by both the "full
Observable" test and the "periodic checkpoint" test below (the
checkpoint is always present in this stream) -- avoids building two
near-identical fixture streams.
Every message is serialized with ``.to_bytes()`` and concatenated with
NO extra framing, exactly matching ``write_capnp_message()``'s "single
stream of consecutive capnp messages" contract (recording.c).
"""
parts: list[bytes] = []
hdr = mod.ReplayHeader.new_message()
hdr.recorderVersion = "1.0"
hdr.recordingStartTs = 1000
hdr.recordingStartWall = "2026-08-05T00:00:00Z"
hdr.schedPolicy = "builtin_fcfs"
hdr.cluster.name = "testcluster"
hdr.cluster.recordingHost = "ctrl-1"
hdr.cluster.init("controllers", 1)
hdr.cluster.controllers[0] = "127.0.0.1:9991"
parts.append(hdr.to_bytes())
snap = mod.SnapshotEntry.new_message()
snap.ts = 1000
snap.data.config.clustername = "testcluster"
snap.data.jobs.ntotal = 1
jobs = snap.data.jobs.init("jobs", 1)
jobs[0].jobid = 501
jobs[0].state = 1
queues = snap.data.init("queues", 1)
queues[0].name = "normal"
queues[0].priority = 10
snap.data.workers.ntotal = 1
workers = snap.data.workers.init("workers", 1)
workers[0].name = "worker1"
snap.data.resources.ntotal = 1
resources = snap.data.resources.init("resources", 1)
resources[0].name = "cpu"
snap.data.resvals.init("vals", 0)
users = snap.data.init("users", 1)
users[0].name = "alice"
projects = snap.data.init("projects", 1)
projects[0].name = "proj1"
projects[0].priority = 5
snap.data.prngSeed = 42
snap.data.internalVersion = 1
parts.append(snap.to_bytes())
ev_push = mod.ReplayEvent.new_message()
ev_push.ts = 1001
jobs_push = ev_push.body.init("jobsPush")
jobs_push.ntotal = 1
jp_jobs = jobs_push.init("jobs", 1)
jp_jobs[0].jobid = 502
parts.append(ev_push.to_bytes())
checkpoint = mod.SnapshotEntry.new_message()
checkpoint.ts = 1005
checkpoint.data.prngSeed = 99
parts.append(checkpoint.to_bytes())
sess = mod.ReplaySession.new_message()
sess.beginRound = 1010
decisions = sess.init("decisions", 2)
decisions[0].ts = 1010
dispatch = decisions[0].body.init("dispatch")
dispatch.alloc.jobid = 502
host_allocs = dispatch.alloc.init("allocs", 1)
host_allocs[0].host = "worker1"
decisions[1].ts = 1011
pend = decisions[1].body.init("pend")
pend.jobid = 503
pend.reason.mainReason = 7
sess.summary.durationNs = 500
sess.summary.nDecisions = 2
sess.summary.nDispatched = 1
sess.summary.nPended = 1
parts.append(sess.to_bytes())
ev_end = mod.ReplayEvent.new_message()
ev_end.ts = 1020
ev_end.body.endSession = None
parts.append(ev_end.to_bytes())
return b"".join(parts)
@pytest.fixture
def full_replay_file(tmp_path: Path) -> Path:
mod = load_schema()
path = tmp_path / "full.replay"
path.write_bytes(_build_full_replay_stream(mod))
return path
def test_dump_lists_header_fields(full_replay_file):
text = scheduler.dump(str(full_replay_file))
assert "=== ReplayHeader ===" in text
assert "recorderVersion: 1.0" in text
assert "schedPolicy: builtin_fcfs" in text
assert "cluster.name: testcluster" in text
assert "127.0.0.1:9991" in text
def test_dump_lists_initial_snapshot_entry(full_replay_file):
text = scheduler.dump(str(full_replay_file))
assert "=== SnapshotEntry (initial) ===" in text
assert "config.clustername: testcluster" in text
assert "jobs: [501]" in text
assert "worker1" in text
assert "proj1" in text
def test_dump_lists_each_event(full_replay_file):
text = scheduler.dump(str(full_replay_file))
assert "=== ReplayEvent ===" in text
assert "type: jobsPush" in text
assert "jobids: [502]" in text
assert "type: endSession" in text
assert "-- end of recording --" in text
def test_dump_lists_session_and_each_decision(full_replay_file):
text = scheduler.dump(str(full_replay_file))
assert "=== ReplaySession ===" in text
assert "beginRound: 1010" in text
assert "decisions: 2" in text
assert "type=dispatch jobid=502 hosts=['worker1']" in text
assert "type=pend jobid=503 mainReason=7" in text
assert "nDispatched=1" in text
assert "nPended=1" in text
def test_dump_periodic_checkpoint_snapshot_still_decodes(full_replay_file):
"""Task 9.3's periodic-checkpoint feature: a mid-stream SnapshotEntry
(not just the initial one at message 2) must still decode correctly.
"""
text = scheduler.dump(str(full_replay_file))
assert text.count("=== SnapshotEntry") == 2
assert "=== SnapshotEntry (initial) ===" in text
assert "=== SnapshotEntry (checkpoint) ===" in text
# the checkpoint carries its own distinct ts/prngSeed, proving it was
# actually decoded (not just a relabeled copy of the initial entry)
assert "ts: 1005" in text
def test_dump_output_ends_with_trailing_newline(full_replay_file):
text = scheduler.dump(str(full_replay_file))
assert text.endswith("\n")
# ---------------------------------------------------------------------------
# dump(): malformed/truncated/empty input handled gracefully (typed error,
# never an uncontrolled crash).
# ---------------------------------------------------------------------------
def test_dump_missing_file_raises_replay_file_error(tmp_path):
missing = tmp_path / "does-not-exist.replay"
with pytest.raises(ReplayFileError) as exc_info:
scheduler.dump(str(missing))
assert exc_info.value.path == str(missing)
def test_dump_empty_file_raises_replay_file_error(tmp_path):
empty = tmp_path / "empty.replay"
empty.write_bytes(b"")
with pytest.raises(ReplayFileError) as exc_info:
scheduler.dump(str(empty))
assert "empty" in exc_info.value.reason
def test_dump_garbage_bytes_raises_replay_file_error_not_crash(tmp_path):
garbage = tmp_path / "garbage.replay"
garbage.write_bytes(b"this is not a capnp message at all, just plain text bytes" * 4)
with pytest.raises(ReplayFileError):
scheduler.dump(str(garbage))
def test_dump_truncated_mid_header_raises_replay_file_error(full_replay_file, tmp_path):
full_bytes = full_replay_file.read_bytes()
truncated = tmp_path / "truncated.replay"
# Cut off partway through the first message -- past its segment table
# (so _peek_message's length calc succeeds) but before all declared
# bytes are actually present.
truncated.write_bytes(full_bytes[:40])
with pytest.raises(ReplayFileError) as exc_info:
scheduler.dump(str(truncated))
assert exc_info.value.reason == "truncated message"
def test_dump_second_message_not_snapshot_entry_raises_replay_file_error(tmp_path):
"""AC 20: the second message MUST be a SnapshotEntry. A file whose
second message is something else (here: a ReplayEvent, wrong shape)
is a format violation dump() must reject, not silently misinterpret --
catches exactly the failure mode this task's own research proved
trial-decoding cannot reliably detect (module docstring in
scheduler.py).
"""
mod = load_schema()
hdr = mod.ReplayHeader.new_message()
hdr.recorderVersion = "1.0"
ev = mod.ReplayEvent.new_message()
ev.ts = 1
ev.body.endSession = None
bad = tmp_path / "bad-second-message.replay"
bad.write_bytes(hdr.to_bytes() + ev.to_bytes())
with pytest.raises(ReplayFileError) as exc_info:
scheduler.dump(str(bad))
assert exc_info.value.reason == "unrecognized message shape"
def test_dump_first_message_not_header_raises_replay_file_error(tmp_path):
"""A file that starts with a SnapshotEntry instead of a ReplayHeader
(wrong shape at message 1) is rejected rather than silently decoded
as a garbage ReplayHeader (empty text fields, no crash -- exactly the
silent-misinterpretation failure mode this task's research ruled out
trial-decoding for; see scheduler.py's dump() section docstring).
"""
mod = load_schema()
snap = mod.SnapshotEntry.new_message()
snap.ts = 1
bad = tmp_path / "bad-first-message.replay"
bad.write_bytes(snap.to_bytes())
with pytest.raises(ReplayFileError) as exc_info:
scheduler.dump(str(bad))
assert exc_info.value.reason == "unrecognized message shape"
def test_dump_raises_on_ambiguous_message_shapes(monkeypatch, full_replay_file):
"""Message-type discrimination relies entirely on ReplayHeader/
SnapshotEntry/ReplayEvent/ReplaySession having pairwise-distinct
(dataWords, pointers) shapes -- verified distinct for the real schema
elsewhere in this file. If a future schema edit ever made two of them
collide, trial-decoding would not raise (this task's own research
finding), so dump() must detect the collision itself and fail loud
with a clear, typed error -- not let an uncaught capnp exception
escape from deep inside from_bytes(). Simulate that collision by
monkeypatching _shape_of() rather than editing the real schema.
"""
real_shape_of = scheduler._shape_of
mod = load_schema()
def _colliding_shape_of(cls):
if cls is mod.SnapshotEntry:
return real_shape_of(mod.ReplaySession)
return real_shape_of(cls)
monkeypatch.setattr(scheduler, "_shape_of", _colliding_shape_of)
with pytest.raises(ReplayFileError) as exc_info:
scheduler.dump(str(full_replay_file))
assert exc_info.value.reason == "ambiguous message shapes"
assert "SnapshotEntry" in str(exc_info.value)
assert "ReplaySession" in str(exc_info.value)