diff --git a/.kiro/specs/scheduler-core/tasks.md b/.kiro/specs/scheduler-core/tasks.md
index 118dd6f4..1e35010a 100644
--- a/.kiro/specs/scheduler-core/tasks.md
+++ b/.kiro/specs/scheduler-core/tasks.md
@@ -273,12 +273,13 @@
   - _Depends: 10.1, 9.4_
   - _Verified: 2026-08-05 — audited task 10.1's own `test_sched_admin_dispatch.c` first: standby rejection and disconnected-scheduler rejection were already fully covered (confirmed unchanged). Two genuine gaps closed: (1) forward timeout — no test previously drove the real 5s `pthread_cond_timedwait` deadline to expiry. Added an `#ifdef XCE_TESTING`-only timeout-override seam in `cluster.c` (`SCHEDADMIN_TIMEOUT_MS` macro + `schedadmin_set_timeout_ms_for_test()`), verified genuinely inert in production builds: `-DXCE_TESTING` is set on exactly the 3 unit-test meson targets, absent from `xce_controller`'s own target and from every project-wide arg; the non-test `#else` branch is a bare `5000L` literal with the test-only global/setter entirely absent from that translation unit; deadline math confirmed correct for both the 5000ms default (behaviorally identical to the original `deadline.tv_sec += 5`) and the test's 200ms override (correct carry logic). `test_schedadmin_forward_timeout` sets a 200ms override, forwards to a real unconnected `sched.ep` with nothing ever replying, brackets the call with real `CLOCK_MONOTONIC` timestamps proving the wait genuinely blocked, and asserts the exact `{status=0, ok=false, error="request timeout"}` shape from requirements.md AC 10 — with the override reset to 5000 strictly before any `TEST_ASSERT` call (verified line-by-line, no `longjmp`-induced test-pollution risk). (2) "Round trip against the real scheduler-side handler" — investigated a literal two-binary integration (controller test invoking the scheduler's real `scheduler_req_handler()` in-process) and judged it architecturally awkward and out of scope (separate meson targets, incompatible fixture types, real symbol-collision risk) with no net correctness gain, since task 9.4's own `test_admin_request.c` already white-box-drives the real scheduler handler rigorously. Instead added two more round-trip outcome-branch tests using the same real `xce_encode`/`xce_decode` wire functions production uses: `test_schedadmin_round_trip_rpc_status_only` (a genuinely new C branch — reply with no `SchedAdminReply` payload) and `test_schedadmin_round_trip_business_error` (same branch as 10.1's existing success test, but proving `ok=false`/error-string field-passthrough fidelity — accurately self-described as payload coverage, not overclaimed as a new branch). Review independently verified every claim against source (confirmed `-DXCE_TESTING`'s exact meson scoping, confirmed sequential single-threaded Unity test execution rules out any cross-test race on the test-only global, confirmed `git diff` exactly matches `git diff -w` for `cluster.c` — zero reformat noise, learning correctly applied from task 10.1's cleanup). Controller suite 3/3 (1093/1093 Unity assertions, up from 1090), scheduler suite 26/26 unaffected, clang-format clean — independently re-verified by me via fresh `meson compile`/`meson test --suite controller`/`meson test --suite scheduler` runs before marking complete._
 
-- [ ] 10.3 Build the xceadmin scheduler recording commands
+- [x] 10.3 Build the xceadmin scheduler recording commands
   - Implement record-start, record-stop, and record-status client functions that connect to a controller, follow redirects to the primary if needed, send the corresponding admin request, and decode the reply
   - Observable: each function returns a decoded reply struct on success and raises a typed error on connection failure, decode failure, or an error reply
   - _Requirements: 5.10_
   - _Depends: 10.1_
   - _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._
 
 - [ ] 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
diff --git a/src/xceadmin/src/xceadmin/scheduler.py b/src/xceadmin/src/xceadmin/scheduler.py
new file mode 100644
index 00000000..8bae7673
--- /dev/null
+++ b/src/xceadmin/src/xceadmin/scheduler.py
@@ -0,0 +1,193 @@
+"""``xceadmin scheduler`` client functions (design.md, ``xceadmin.scheduler``;
+Requirements 5.10, 5.20).
+
+This module is the shared home for the scheduler recording-control client
+functions (``record_start``/``record_stop``/``record_status``, task 10.3)
+AND the local ``.replay`` file decoder (``dump()``, task 10.4, NOT built
+here) -- design.md's File Structure Plan places both in this one file,
+mirroring ``push.py``'s split (all business logic here, ``cli.py`` stays a
+thin ``click`` layer -- task 10.5, also not built here).
+
+``record_start``/``record_stop``/``record_status`` mirror ``push.py``'s
+``push_config()`` pipeline exactly: build request bytes with the real
+``controller.capnp`` schema (``xcecommon.schema.load_schema()``) -> connect
++ follow redirects to the primary (``xceadmin.discovery.follow_redirects``,
+Requirement 3.1-3.4 -- the scheduler is a singleton attached to the
+primary controller, so these commands must reach it the same way ``config
+push`` reaches the primary) -> decode the ``Request{opcode: reply,
+body.replymsg: RequestReply}`` envelope (``xceadmin.discovery.decode_reply``)
+-> extract the ``SchedAdminReply`` payload. All three actions share one
+``SchedAdminRequest``/``SchedAdminReply`` wire shape
+(``controller.capnp``'s ``schedadminmsg``/``schedadminreplymsg`` union
+members, added in task 7.4) with only ``action`` (and, for ``recordStart``
+only, ``recordFile``) varying -- so all three go through one shared
+``_send_admin_request()`` helper rather than three near-identical
+functions.
+
+**Error-vs-return convention -- a deliberate divergence from ``push.py``**:
+``push_config()`` returns an application-level failure (``status != 0`` or
+a non-empty ``UpgradeConfigReply.error``) as data on ``PushResult`` for the
+CLI layer to interpret (design.md's `xceadmin.push` Boundary Commitments;
+`cli.py` checks ``result.error or result.acked_count <= 0`` itself). Task
+10.3's Observable is written differently and explicitly: "raises a typed
+error on connection failure, decode failure, or **an error reply**" --
+naming the business-level ``ok=false`` case as a THIRD raising failure
+mode, on equal footing with connection/decode failures, not a fourth
+"return it and let the caller decide" case. This module follows that
+literal instruction rather than ``push_config``'s precedent: a
+``SchedAdminReply{ok=false, error=...}`` raises
+:class:`~xcecommon.errors.SchedAdminError` (new, ``xcecommon.errors``)
+carrying the full reply state (``error``, ``recording``, ``file``,
+``event_count``, ``started_at``) so a caller that wants to report it (e.g.
+"already recording -- current file: X") does not need a second
+``record_status()`` round-trip. Connection failures
+(:class:`~xcecommon.errors.ConnectionTimeoutError` /
+:class:`~xcecommon.errors.NoPrimaryError` /
+:class:`~xcecommon.errors.TooManyRedirectsError`) and decode failures
+(:class:`~xcecommon.errors.ReplyDecodeError`) already raise by construction
+of ``follow_redirects``/``decode_reply`` -- reused as-is, not reimplemented.
+
+On success, the decoded ``SchedAdminReply`` is returned as
+:class:`RecordResult` -- ``recording``/``file``/``event_count``/
+``started_at`` only (``ok``/``error`` are omitted: since a
+``ok=false`` reply now always raises, they would be trivially
+``True``/``""`` on every successful return, matching ``DryRunResult``'s
+precedent of re-deriving semantic fields rather than mirroring the raw
+wire struct verbatim).
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from xcecommon import wire
+from xcecommon.errors import ReplyDecodeError, SchedAdminError
+from xcecommon.schema import load_schema
+from xceadmin.discovery import decode_reply, follow_redirects
+
+_DEFAULT_PORT = 9991
+
+
+@dataclass(frozen=True)
+class RecordResult:
+    recording: bool
+    file: str
+    event_count: int
+    started_at: int
+
+
+def _split_controller(controller: str) -> tuple[str, int]:
+    """``"host:port"`` -> ``(host, port)``; a bare host uses the default
+    port 9991 -- same convention as ``push.py``'s own ``_split_controller``
+    (each resource module owns this trivial parsing rather than sharing a
+    private helper across modules, matching this package's existing
+    per-module-self-contained precedent).
+    """
+    if ":" in controller:
+        host, port_str = controller.rsplit(":", 1)
+        return host, int(port_str)
+    return controller, _DEFAULT_PORT
+
+
+def _send_via(request_bytes: bytes):
+    def _send(host: str, port: int) -> bytes:
+        sock = wire.connect(host, port)
+        try:
+            return wire.send_request(sock, request_bytes)
+        finally:
+            sock.close()
+
+    return _send
+
+
+def _build_request(action: str, record_file: str = "") -> bytes:
+    mod = load_schema()
+    request = mod.Request.new_message()
+    request.opcode = "schedAdmin"
+    request.seq = 1
+    admin = request.body.init("schedadminmsg")
+    admin.action = action
+    admin.recordFile = record_file
+    return request.to_bytes()
+
+
+def _send_admin_request(controller: str, action: str, record_file: str = "") -> RecordResult:
+    request_bytes = _build_request(action, record_file)
+
+    host, port = _split_controller(controller)
+    _, _, reply_bytes = follow_redirects(host, port, _send_via(request_bytes))
+
+    with decode_reply(reply_bytes) as reply:
+        if reply.status != 0:
+            # Defensive, mirrors push_config()'s own status-first check:
+            # a non-zero RequestReply.status means reply.data may not even
+            # be the schedadminreplymsg variant (e.g. a bare debugmsg), so
+            # this is reported before ever inspecting reply.data.which().
+            reason = f"controller returned status {reply.status}"
+            raise SchedAdminError(
+                f"scheduler admin request failed: {reason}",
+                error=reason,
+                recording=False,
+                file="",
+                event_count=0,
+                started_at=0,
+            )
+
+        if reply.data.which() != "schedadminreplymsg":
+            reason = f"unexpected reply variant {reply.data.which()!r}, expected schedadminreplymsg"
+            raise ReplyDecodeError(f"invalid reply: {reason}", reason)
+
+        admin_reply = reply.data.schedadminreplymsg
+        if not admin_reply.ok:
+            raise SchedAdminError(
+                f"scheduler admin request failed: {admin_reply.error}",
+                error=admin_reply.error,
+                recording=admin_reply.recording,
+                file=admin_reply.file,
+                event_count=admin_reply.eventCount,
+                started_at=admin_reply.startedAt,
+            )
+
+        return RecordResult(
+            recording=admin_reply.recording,
+            file=admin_reply.file,
+            event_count=admin_reply.eventCount,
+            started_at=admin_reply.startedAt,
+        )
+
+
+def record_start(controller: str, filename: str | None = None) -> RecordResult:
+    """Start scheduler recording (``xceadmin scheduler record start
+    [FILE]``): ``SchedAdminRequest{action=recordStart, recordFile=filename
+    or ""}`` (requirements.md AC 10 -- empty ``recordFile`` means the
+    scheduler auto-generates a name).
+
+    Raises :class:`~xcecommon.errors.SchedAdminError` if already recording
+    (module docstring); connection/decode failures raise their own
+    existing typed errors (``xceadmin.discovery``).
+    """
+    return _send_admin_request(controller, "recordStart", filename or "")
+
+
+def record_stop(controller: str) -> RecordResult:
+    """Stop scheduler recording (``xceadmin scheduler record stop``):
+    ``SchedAdminRequest{action=recordStop}``.
+
+    Raises :class:`~xcecommon.errors.SchedAdminError` if not currently
+    recording (module docstring).
+    """
+    return _send_admin_request(controller, "recordStop")
+
+
+def record_status(controller: str) -> RecordResult:
+    """Query scheduler recording status (``xceadmin scheduler record
+    status``): ``SchedAdminRequest{action=recordStatus}``.
+
+    ``recordStatus`` always replies ``ok=true`` (requirements.md's Reply
+    field semantics table has no error row for this action), so this only
+    raises for connection/decode failures in practice -- kept behind the
+    same shared helper as ``record_start``/``record_stop`` rather than a
+    special-cased fourth code path, matching the "Reply field semantics by
+    action" table's own framing of all three actions as one contract.
+    """
+    return _send_admin_request(controller, "recordStatus")
diff --git a/src/xceadmin/src/xcecommon/errors.py b/src/xceadmin/src/xcecommon/errors.py
index 4c27c7ce..6a5f6b2b 100644
--- a/src/xceadmin/src/xcecommon/errors.py
+++ b/src/xceadmin/src/xcecommon/errors.py
@@ -127,3 +127,42 @@ class NoPrimaryError(XceAdminError):
         super().__init__(message)
         self.host = host
         self.port = port
+
+
+class SchedAdminError(XceAdminError):
+    """A ``SchedAdminRequest`` (scheduler recording control) round-tripped
+    successfully but the reply itself reports failure --
+    ``SchedAdminReply{ok=false, error=...}`` (Requirement 5.10; declared
+    here, raised by a later task -- see ``xceadmin.scheduler``).
+
+    Unlike ``config push``'s ``PushResult``/``DryRunResult`` (which return
+    an application-level failure as data for the caller to inspect), the
+    scheduler recording commands' task text is explicit: "raises a typed
+    error on connection failure, decode failure, or an error reply" --
+    three failure modes treated uniformly. This is a deliberate,
+    documented divergence from ``push_config``'s convention, not an
+    oversight (see ``xceadmin.scheduler`` module docstring).
+
+    Carries the full reply state alongside ``error``, not just the reason
+    string -- requirements.md's "Reply field semantics by action" table
+    shows a business-level failure (e.g. "already recording") still
+    reports meaningful ``recording``/``file`` state, so a caller that
+    wants to display it (e.g. "already recording -- current file: X")
+    does not have to re-query ``record status`` separately.
+    """
+
+    def __init__(
+        self,
+        message: str,
+        error: str,
+        recording: bool,
+        file: str,
+        event_count: int,
+        started_at: int,
+    ) -> None:
+        super().__init__(message)
+        self.error = error
+        self.recording = recording
+        self.file = file
+        self.event_count = event_count
+        self.started_at = started_at
diff --git a/src/xceadmin/tests/test_errors.py b/src/xceadmin/tests/test_errors.py
index 1d0595d9..c5bd2517 100644
--- a/src/xceadmin/tests/test_errors.py
+++ b/src/xceadmin/tests/test_errors.py
@@ -21,6 +21,7 @@ from xcecommon.errors import (
     NoPrimaryError,
     ReplyDecodeError,
     ReservedNameError,
+    SchedAdminError,
     SchemaNotFoundError,
     TooManyRedirectsError,
     ValidationError,
@@ -36,6 +37,7 @@ ALL_SUBCLASSES = [
     ReplyDecodeError,
     TooManyRedirectsError,
     NoPrimaryError,
+    SchedAdminError,
 ]
 
 
@@ -171,6 +173,31 @@ def test_no_primary_error_carries_host_and_port() -> None:
     assert caught.port == 9991
 
 
+def test_sched_admin_error_carries_error_and_reply_state() -> None:
+    """Requirement 5.10 (declared here, raised by a later task): a
+    ``SchedAdminReply{ok=false, error=...}`` is surfaced as a typed error
+    carrying the reason plus the reply's recording/file/event/started-at
+    state (e.g. "already recording" still reports the current file)."""
+    with pytest.raises(XceAdminError) as exc_info:
+        raise SchedAdminError(
+            "scheduler admin request failed: already recording",
+            error="already recording",
+            recording=True,
+            file="/var/log/xce/current.replay",
+            event_count=42,
+            started_at=1234567890,
+        )
+
+    caught = exc_info.value
+    assert isinstance(caught, SchedAdminError)
+    assert caught.message == "scheduler admin request failed: already recording"
+    assert caught.error == "already recording"
+    assert caught.recording is True
+    assert caught.file == "/var/log/xce/current.replay"
+    assert caught.event_count == 42
+    assert caught.started_at == 1234567890
+
+
 def test_catching_base_type_recovers_message_for_every_subclass() -> None:
     """The entire point of the shared exception contract (design.md,
     xceadmin.cli Implementation Note): a single `except XceAdminError as e`
@@ -186,6 +213,12 @@ def test_catching_base_type_recovers_message_for_every_subclass() -> None:
         (ReplyDecodeError("msg-f", reason="r"), "msg-f"),
         (TooManyRedirectsError("msg-g", hops=3), "msg-g"),
         (NoPrimaryError("msg-h", host="h", port=1), "msg-h"),
+        (
+            SchedAdminError(
+                "msg-i", error="e", recording=False, file="", event_count=0, started_at=0
+            ),
+            "msg-i",
+        ),
     ]
 
     for exc, expected_message in cases:
diff --git a/src/xceadmin/tests/test_scheduler.py b/src/xceadmin/tests/test_scheduler.py
new file mode 100644
index 00000000..209e8630
--- /dev/null
+++ b/src/xceadmin/tests/test_scheduler.py
@@ -0,0 +1,410 @@
+"""Mock-controller tests for ``xceadmin.scheduler``'s record-control client
+functions (design.md, ``xceadmin.scheduler``; Requirements 5.10).
+
+Mirrors ``test_config_push_mock.py``'s mock TLS server pattern exactly
+(same ``mock_controller``/``wrap_reply``/``tls_certs`` fixtures from
+``conftest.py``) -- ``record_start``/``record_stop``/``record_status``
+have no live controller counterpart to test against yet (task 10.1's
+controller-side dispatch exists, but there is no running cluster in this
+test environment), so these are tested against a real local TLS mock
+server built from the actual bundled schema, exactly as
+``test_config_push_mock.py`` does for ``push_config``/``probe_config``.
+
+Only ``record_start``/``record_stop``/``record_status`` (task 10.3) are
+covered here. ``dump()`` (task 10.4, not yet built) will get its own
+golden-file tests appended to this file later (design.md's File Structure
+Plan places both in one ``test_scheduler.py``) -- not duplicated or
+anticipated here.
+"""
+
+from __future__ import annotations
+
+import ssl
+import socket
+import threading
+from pathlib import Path
+
+import pytest
+
+from xcecommon import wire
+from xcecommon.errors import NoPrimaryError, ReplyDecodeError, SchedAdminError
+from xcecommon.schema import load_schema
+from xceadmin import scheduler
+
+
+def _server_context(cert_file: Path, key_file: Path) -> ssl.SSLContext:
+    context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
+    context.load_cert_chain(certfile=str(cert_file), keyfile=str(key_file))
+    return context
+
+
+class _MockController:
+    """A one-shot TLS server: accepts one connection, hands the raw
+    accepted TLS socket to ``handler``, then closes. Identical to
+    ``test_config_push_mock.py``'s own ``_MockController`` -- not shared
+    across test files since pytest fixtures/helpers are conventionally
+    module-local in this test suite (each of ``test_config_push_mock.py``,
+    this file, etc. is self-contained).
+    """
+
+    def __init__(self, handler, cert_file: Path, key_file: Path):
+        self._handler = handler
+        self._cert_file = cert_file
+        self._key_file = key_file
+        self._listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        self._listener.bind(("127.0.0.1", 0))
+        self._listener.listen(1)
+        self.port = self._listener.getsockname()[1]
+        self._thread = threading.Thread(target=self._run, daemon=True)
+        self._thread.start()
+
+    def _run(self) -> None:
+        try:
+            raw_conn, _ = self._listener.accept()
+        except OSError:
+            return
+        try:
+            tls_conn = _server_context(self._cert_file, self._key_file).wrap_socket(
+                raw_conn, server_side=True
+            )
+        except (ssl.SSLError, OSError):
+            raw_conn.close()
+            return
+        try:
+            self._handler(tls_conn)
+        except OSError:
+            pass
+        finally:
+            tls_conn.close()
+
+    def close(self) -> None:
+        self._listener.close()
+        self._thread.join(timeout=5)
+
+
+@pytest.fixture
+def mock_controller(tls_certs):
+    servers: list[_MockController] = []
+
+    def _make(handler) -> _MockController:
+        srv = _MockController(handler, tls_certs.cert_file, tls_certs.key_file)
+        servers.append(srv)
+        return srv
+
+    yield _make
+    for srv in servers:
+        srv.close()
+
+
+# ---------------------------------------------------------------------------
+# record_start(): happy path, redirect-following
+# ---------------------------------------------------------------------------
+
+
+def test_record_start_auto_generated_filename(mock_controller, wrap_reply):
+    mod = load_schema()
+    received = {}
+
+    def _handler(conn):
+        request_bytes = wire.recv_frame(conn)
+        with mod.Request.from_bytes(request_bytes) as req:
+            received["opcode"] = req.opcode
+            received["action"] = req.body.schedadminmsg.action
+            received["recordFile"] = req.body.schedadminmsg.recordFile
+
+        reply = mod.RequestReply.new_message()
+        reply.status = 0
+        admin_reply = reply.data.init("schedadminreplymsg")
+        admin_reply.ok = True
+        admin_reply.error = ""
+        admin_reply.recording = True
+        admin_reply.file = "/var/log/xce/scheduler-replay-auto.replay"
+        admin_reply.eventCount = 0
+        admin_reply.startedAt = 1000
+        wire.send_frame(conn, wrap_reply(reply))
+
+    srv = mock_controller(_handler)
+
+    result = scheduler.record_start(f"127.0.0.1:{srv.port}")
+
+    assert received["opcode"] == "schedAdmin"
+    assert received["action"] == "recordStart"
+    assert received["recordFile"] == ""
+    assert result.recording is True
+    assert result.file == "/var/log/xce/scheduler-replay-auto.replay"
+    assert result.event_count == 0
+    assert result.started_at == 1000
+
+
+def test_record_start_explicit_filename_sent_on_wire(mock_controller, wrap_reply):
+    mod = load_schema()
+    received = {}
+
+    def _handler(conn):
+        request_bytes = wire.recv_frame(conn)
+        with mod.Request.from_bytes(request_bytes) as req:
+            received["recordFile"] = req.body.schedadminmsg.recordFile
+
+        reply = mod.RequestReply.new_message()
+        reply.status = 0
+        admin_reply = reply.data.init("schedadminreplymsg")
+        admin_reply.ok = True
+        admin_reply.error = ""
+        admin_reply.recording = True
+        admin_reply.file = "myfile.replay"
+        admin_reply.eventCount = 0
+        admin_reply.startedAt = 2000
+        wire.send_frame(conn, wrap_reply(reply))
+
+    srv = mock_controller(_handler)
+
+    result = scheduler.record_start(f"127.0.0.1:{srv.port}", filename="myfile.replay")
+
+    assert received["recordFile"] == "myfile.replay"
+    assert result.file == "myfile.replay"
+
+
+def test_record_start_follows_one_redirect_to_primary(mock_controller, wrap_reply):
+    mod = load_schema()
+
+    def _standby_handler(conn):
+        wire.recv_frame(conn)
+        reply = mod.RequestReply.new_message()
+        reply.status = 0
+        reply.redirectHost = "127.0.0.1"
+        reply.redirectPort = _standby_handler.primary_port
+        reply.data.nodata = None
+        wire.send_frame(conn, wrap_reply(reply))
+
+    def _primary_handler(conn):
+        wire.recv_frame(conn)
+        reply = mod.RequestReply.new_message()
+        reply.status = 0
+        admin_reply = reply.data.init("schedadminreplymsg")
+        admin_reply.ok = True
+        admin_reply.error = ""
+        admin_reply.recording = True
+        admin_reply.file = "on-primary.replay"
+        admin_reply.eventCount = 0
+        admin_reply.startedAt = 3000
+        wire.send_frame(conn, wrap_reply(reply))
+
+    primary_srv = mock_controller(_primary_handler)
+    _standby_handler.primary_port = primary_srv.port
+    standby_srv = mock_controller(_standby_handler)
+
+    result = scheduler.record_start(f"127.0.0.1:{standby_srv.port}")
+
+    assert result.file == "on-primary.replay"
+
+
+# ---------------------------------------------------------------------------
+# record_stop() / record_status(): happy path
+# ---------------------------------------------------------------------------
+
+
+def test_record_stop_successful(mock_controller, wrap_reply):
+    mod = load_schema()
+    received = {}
+
+    def _handler(conn):
+        request_bytes = wire.recv_frame(conn)
+        with mod.Request.from_bytes(request_bytes) as req:
+            received["action"] = req.body.schedadminmsg.action
+
+        reply = mod.RequestReply.new_message()
+        reply.status = 0
+        admin_reply = reply.data.init("schedadminreplymsg")
+        admin_reply.ok = True
+        admin_reply.error = ""
+        admin_reply.recording = False
+        admin_reply.file = "final.replay"
+        admin_reply.eventCount = 14273
+        admin_reply.startedAt = 1000
+        wire.send_frame(conn, wrap_reply(reply))
+
+    srv = mock_controller(_handler)
+
+    result = scheduler.record_stop(f"127.0.0.1:{srv.port}")
+
+    assert received["action"] == "recordStop"
+    assert result.recording is False
+    assert result.event_count == 14273
+    assert result.file == "final.replay"
+
+
+def test_record_status_successful(mock_controller, wrap_reply):
+    mod = load_schema()
+    received = {}
+
+    def _handler(conn):
+        request_bytes = wire.recv_frame(conn)
+        with mod.Request.from_bytes(request_bytes) as req:
+            received["action"] = req.body.schedadminmsg.action
+
+        reply = mod.RequestReply.new_message()
+        reply.status = 0
+        admin_reply = reply.data.init("schedadminreplymsg")
+        admin_reply.ok = True
+        admin_reply.error = ""
+        admin_reply.recording = True
+        admin_reply.file = "current.replay"
+        admin_reply.eventCount = 500
+        admin_reply.startedAt = 4000
+        wire.send_frame(conn, wrap_reply(reply))
+
+    srv = mock_controller(_handler)
+
+    result = scheduler.record_status(f"127.0.0.1:{srv.port}")
+
+    assert received["action"] == "recordStatus"
+    assert result.recording is True
+    assert result.event_count == 500
+
+
+# ---------------------------------------------------------------------------
+# ok=false replies raise SchedAdminError (task 10.3's Observable: "raises a
+# typed error on ... an error reply")
+# ---------------------------------------------------------------------------
+
+
+def test_record_start_already_recording_raises_sched_admin_error(mock_controller, wrap_reply):
+    mod = load_schema()
+
+    def _handler(conn):
+        wire.recv_frame(conn)
+        reply = mod.RequestReply.new_message()
+        reply.status = 0
+        admin_reply = reply.data.init("schedadminreplymsg")
+        admin_reply.ok = False
+        admin_reply.error = "already recording"
+        admin_reply.recording = True
+        admin_reply.file = "current.replay"
+        admin_reply.eventCount = 7
+        admin_reply.startedAt = 5000
+        wire.send_frame(conn, wrap_reply(reply))
+
+    srv = mock_controller(_handler)
+
+    with pytest.raises(SchedAdminError) as exc_info:
+        scheduler.record_start(f"127.0.0.1:{srv.port}")
+
+    caught = exc_info.value
+    assert caught.error == "already recording"
+    assert caught.recording is True
+    assert caught.file == "current.replay"
+    assert caught.event_count == 7
+    assert caught.started_at == 5000
+
+
+def test_record_stop_not_recording_raises_sched_admin_error(mock_controller, wrap_reply):
+    mod = load_schema()
+
+    def _handler(conn):
+        wire.recv_frame(conn)
+        reply = mod.RequestReply.new_message()
+        reply.status = 0
+        admin_reply = reply.data.init("schedadminreplymsg")
+        admin_reply.ok = False
+        admin_reply.error = "not recording"
+        admin_reply.recording = False
+        admin_reply.file = ""
+        admin_reply.eventCount = 0
+        admin_reply.startedAt = 0
+        wire.send_frame(conn, wrap_reply(reply))
+
+    srv = mock_controller(_handler)
+
+    with pytest.raises(SchedAdminError) as exc_info:
+        scheduler.record_stop(f"127.0.0.1:{srv.port}")
+
+    assert exc_info.value.error == "not recording"
+    assert exc_info.value.recording is False
+
+
+@pytest.mark.parametrize(
+    "error_reason", ["scheduler not connected", "request timeout", "invalid record file name"]
+)
+def test_record_start_error_codes_raise_sched_admin_error(mock_controller, wrap_reply, error_reason):
+    """requirements.md's "Error codes reference" table -- every listed
+    ``SchedAdminReply{ok=false, error=...}`` reason raises the same typed
+    error, regardless of which specific reason string the reply carries.
+    """
+    mod = load_schema()
+
+    def _handler(conn):
+        wire.recv_frame(conn)
+        reply = mod.RequestReply.new_message()
+        reply.status = 0
+        admin_reply = reply.data.init("schedadminreplymsg")
+        admin_reply.ok = False
+        admin_reply.error = error_reason
+        admin_reply.recording = False
+        admin_reply.file = ""
+        admin_reply.eventCount = 0
+        admin_reply.startedAt = 0
+        wire.send_frame(conn, wrap_reply(reply))
+
+    srv = mock_controller(_handler)
+
+    with pytest.raises(SchedAdminError) as exc_info:
+        scheduler.record_start(f"127.0.0.1:{srv.port}")
+
+    assert exc_info.value.error == error_reason
+
+
+# ---------------------------------------------------------------------------
+# Decode failure (wrong reply variant) raises ReplyDecodeError
+# ---------------------------------------------------------------------------
+
+
+def test_record_status_wrong_reply_variant_raises_reply_decode_error(mock_controller, wrap_reply):
+    mod = load_schema()
+
+    def _handler(conn):
+        wire.recv_frame(conn)
+        reply = mod.RequestReply.new_message()
+        reply.status = 0
+        reply.data.nodata = None  # not schedadminreplymsg
+        wire.send_frame(conn, wrap_reply(reply))
+
+    srv = mock_controller(_handler)
+
+    with pytest.raises(ReplyDecodeError):
+        scheduler.record_status(f"127.0.0.1:{srv.port}")
+
+
+# ---------------------------------------------------------------------------
+# Connection failure raises NoPrimaryError (end-to-end composition with
+# xceadmin.discovery, mirrors test_config_push_mock.py's own equivalent
+# test for push_config)
+# ---------------------------------------------------------------------------
+
+
+def test_record_start_propagates_no_primary_error_on_connection_timeout(monkeypatch):
+    from xcecommon.errors import ConnectionTimeoutError
+
+    def _fake_connect(host, port, timeout_s=30.0):
+        raise ConnectionTimeoutError(f"timeout connecting to {host}:{port}", host, port)
+
+    monkeypatch.setattr(wire, "connect", _fake_connect)
+
+    with pytest.raises(NoPrimaryError):
+        scheduler.record_start("127.0.0.1:9991")
+
+
+# ---------------------------------------------------------------------------
+# _split_controller(): same default-port convention as push.py
+# ---------------------------------------------------------------------------
+
+
+def test_record_default_port_used_when_omitted():
+    host, port = scheduler._split_controller("somehost")
+    assert host == "somehost"
+    assert port == 9991
+
+
+def test_record_explicit_port_parsed():
+    host, port = scheduler._split_controller("somehost:1234")
+    assert host == "somehost"
+    assert port == 1234