"""Mock-controller tests for ``xceadmin.scheduler``'s record-control client
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
(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``.

``dump()`` tests (bottom of this file) are a completely separate concern
-- no network, no mock controller. No pre-existing ``.replay`` fixture
file exists anywhere in this repo (confirmed by a repo-wide search before
writing these tests) and the C-side simulation driver that would consume
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

import ssl
import socket
import threading
from pathlib import Path

import pytest

from xcecommon import wire
from xcecommon.errors import NoPrimaryError, ReplayFileError, ReplyDecodeError, SchedAdminError
from xcecommon.schema import load_schema
from xceadmin import discovery, 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",
    [
        "invalid record file name",
        "record directory not found",
        "permission denied accessing record file",
    ],
)
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=...}`` PERMANENT reason (driver.c's
    ``validation_error_text()`` filename/path-rejection strings) raises the
    same typed error immediately, on the first attempt, regardless of which
    specific reason string the reply carries. Uses the one-shot
    ``mock_controller`` fixture deliberately (not ``sequential_mock_
    controller``): if bug-1715's new admin-reason retry logic ever
    incorrectly treated one of these as transient, a second connection
    attempt against this one-shot server would fail with a connection
    error instead of the expected ``SchedAdminError``, independently
    proving single-attempt behavior.

    bug-1715: "scheduler not connected" and "request timeout" used to be
    tested here too, but are now confirmed-transient reasons (module
    comment on ``scheduler._TRANSIENT_ADMIN_REASONS``) and are retried, not
    raised immediately -- moved to the dedicated retry tests below.
    """
    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")


# ---------------------------------------------------------------------------
# bug-1713: E_XCE_NOMASTER (-10011) / E_XCE_NOT_READY (-10013) are transient
# "no settled primary right now" conditions -- retried with a bounded
# budget instead of failing immediately, unlike every other nonzero status.
# E_XCE_STANDBY (-10014) is deliberately NOT part of this: it is resolved
# purely by follow_redirects()'s existing redirect-following (bug-1701),
# regression-locked at the bottom of this section.
#
# Uses a dedicated multi-connection mock server (_SequentialMockController)
# rather than the module's single-shot ``mock_controller`` fixture:
# scheduler.py's retry loop re-runs follow_redirects() from scratch on each
# attempt (a fresh TCP connection), not a resend over one held connection,
# so these tests must let the mock server accept more than one connection
# on the same port.
# ---------------------------------------------------------------------------


class _SequentialMockController:
    """Like ``_MockController`` above, but accepts exactly ``len(handlers)``
    connections in sequence on the same port (not just one), handing each
    accepted TLS connection to the next handler in ``handlers`` in order --
    module section docstring above explains why this is needed for
    bug-1713's retry tests specifically, instead of reusing
    ``_MockController``. Callers pass exactly as many handlers as
    connections they expect (proven independently via
    ``connection_count``) -- the accept loop is a plain, finite ``for``
    over ``handlers``, so it never blocks waiting for a connection beyond
    what the test itself expects, and closes/joins promptly in ``close()``
    even when a test ends early. A short accept timeout is still set as a
    safety net: if a regression ever made the client under test connect
    MORE times than the test provisioned handlers for, this server exits
    its loop and closes within one timeout window rather than blocking
    ``close()``'s ``join()`` for its full grace period.
    """

    _ACCEPT_TIMEOUT_S = 2.0

    def __init__(self, handlers, cert_file: Path, key_file: Path):
        self._handlers = list(handlers)
        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(5)
        self._listener.settimeout(self._ACCEPT_TIMEOUT_S)
        self.port = self._listener.getsockname()[1]
        self.connection_count = 0
        self._thread = threading.Thread(target=self._run, daemon=True)
        self._thread.start()

    def _run(self) -> None:
        for handler in self._handlers:
            try:
                raw_conn, _ = self._listener.accept()
            except OSError:
                return
            self.connection_count += 1
            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()
                continue
            try:
                handler(tls_conn)
            except OSError:
                pass
            finally:
                tls_conn.close()

    def close(self) -> None:
        self._listener.close()
        self._thread.join(timeout=self._ACCEPT_TIMEOUT_S + 1)


@pytest.fixture
def sequential_mock_controller(tls_certs):
    servers: list[_SequentialMockController] = []

    def _make(handlers) -> _SequentialMockController:
        srv = _SequentialMockController(handlers, tls_certs.cert_file, tls_certs.key_file)
        servers.append(srv)
        return srv

    yield _make
    for srv in servers:
        srv.close()


@pytest.mark.parametrize(
    "transient_status",
    [scheduler._E_XCE_NOT_READY, scheduler._E_XCE_NOMASTER],
    ids=["not_ready", "nomaster"],
)
def test_record_status_retries_on_transient_status_then_succeeds(
    sequential_mock_controller, wrap_reply, monkeypatch, transient_status
):
    """bug-1713's core proof: a mocked server returning a transient status
    (E_XCE_NOT_READY or E_XCE_NOMASTER) twice, then succeeding on the 3rd
    attempt, results in the CLIENT EVENTUALLY SUCCEEDING -- not an
    immediate raise. FAILS against the pre-fix code (which raises
    SchedAdminError on the very first reply)."""
    mod = load_schema()
    sleep_calls: list[float] = []
    monkeypatch.setattr(scheduler.time, "sleep", lambda s: sleep_calls.append(s))

    def _not_ready_handler(conn):
        wire.recv_frame(conn)
        reply = mod.RequestReply.new_message()
        reply.status = transient_status
        reply.data.nodata = None
        wire.send_frame(conn, wrap_reply(reply))

    def _success_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 = "settled.replay"
        admin_reply.eventCount = 0
        admin_reply.startedAt = 6000
        wire.send_frame(conn, wrap_reply(reply))

    srv = sequential_mock_controller([_not_ready_handler, _not_ready_handler, _success_handler])

    result = scheduler.record_status(f"127.0.0.1:{srv.port}")

    assert result.file == "settled.replay"
    assert srv.connection_count == 3
    # Slept between attempt 1->2 and 2->3, but not after the final success.
    assert sleep_calls == [scheduler._TRANSIENT_RETRY_DELAY_S] * 2


def test_record_status_retry_exhaustion_raises_distinguishable_error(
    sequential_mock_controller, wrap_reply, monkeypatch
):
    """A mocked server that ALWAYS returns E_XCE_NOT_READY (never
    recovers) must make the client eventually give up with a clear,
    distinguishable SchedAdminError after the retry budget is exhausted --
    not a silent hang or an infinite loop. ``time.sleep`` is monkeypatched
    so this test does not actually wait out the real retry delay."""
    mod = load_schema()
    sleep_calls: list[float] = []
    monkeypatch.setattr(scheduler.time, "sleep", lambda s: sleep_calls.append(s))

    def _not_ready_handler(conn):
        wire.recv_frame(conn)
        reply = mod.RequestReply.new_message()
        reply.status = scheduler._E_XCE_NOT_READY
        reply.data.nodata = None
        wire.send_frame(conn, wrap_reply(reply))

    # Exactly the retry budget's worth of handlers: connection_count is
    # asserted below to equal _MAX_TRANSIENT_ATTEMPTS exactly, proving the
    # client stops on its OWN bounded budget -- not a weaker "it didn't
    # exceed however many the mock happened to provide" proof.
    srv = sequential_mock_controller([_not_ready_handler] * scheduler._MAX_TRANSIENT_ATTEMPTS)

    with pytest.raises(SchedAdminError) as exc_info:
        scheduler.record_status(f"127.0.0.1:{srv.port}")

    assert srv.connection_count == scheduler._MAX_TRANSIENT_ATTEMPTS
    assert len(sleep_calls) == scheduler._MAX_TRANSIENT_ATTEMPTS - 1
    # Distinguishable from an immediate, non-retried hard failure: the
    # message names both the raw status and that retries were exhausted.
    assert str(scheduler._E_XCE_NOT_READY) in exc_info.value.error
    assert "giving up" in exc_info.value.error
    assert f"{scheduler._MAX_TRANSIENT_ATTEMPTS} attempts" in exc_info.value.error


def test_record_status_retries_on_no_primary_error_then_succeeds(monkeypatch, wrap_reply):
    """Right after a primary is killed, every configured candidate can
    still redirect to the now-dead old primary (a standby hasn't
    necessarily noticed within the heartbeat timeout yet), so
    follow_redirects_any exhausts the whole candidate list and raises
    NoPrimaryError even though the cluster is only mid-election, not
    genuinely unreachable. This must be retried using NoPrimaryError's own
    (longer) budget/delay, _MAX_NO_PRIMARY_ATTEMPTS/
    _NO_PRIMARY_RETRY_DELAY_S -- not treated as an immediate hard failure,
    and not sharing _TRANSIENT_STATUSES' smaller budget. FAILS against the
    pre-fix code (which let NoPrimaryError propagate out of the retry loop
    on the very first attempt)."""
    mod = load_schema()
    sleep_calls: list[float] = []
    monkeypatch.setattr(scheduler.time, "sleep", lambda s: sleep_calls.append(s))

    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 = "settled.replay"
    admin_reply.eventCount = 0
    admin_reply.startedAt = 6000
    success_reply_bytes = wrap_reply(reply)

    calls = {"n": 0}

    def fake_follow_redirects_any(candidates, send_fn):
        calls["n"] += 1
        if calls["n"] < 3:
            raise NoPrimaryError("no primary available: none of the configured controllers responded", "host1", 9991)
        return "host2", 9992, success_reply_bytes

    monkeypatch.setattr(scheduler, "follow_redirects_any", fake_follow_redirects_any)

    result = scheduler.record_status("host1:9991,host2:9992,host3:9993")

    assert result.file == "settled.replay"
    assert calls["n"] == 3
    assert sleep_calls == [scheduler._NO_PRIMARY_RETRY_DELAY_S] * 2


def test_record_status_no_primary_error_retry_exhaustion_propagates(monkeypatch):
    """If NoPrimaryError persists across the whole retry budget (the
    cluster genuinely never settles, or every candidate really is down),
    it must still propagate as NoPrimaryError -- not be silently
    swallowed or converted into a less-informative error -- after exactly
    _MAX_NO_PRIMARY_ATTEMPTS attempts (its own, longer budget -- NOT
    _MAX_TRANSIENT_ATTEMPTS), matching the reply-status exhaustion test's
    own proof shape."""
    sleep_calls: list[float] = []
    monkeypatch.setattr(scheduler.time, "sleep", lambda s: sleep_calls.append(s))

    calls = {"n": 0}

    def fake_follow_redirects_any(candidates, send_fn):
        calls["n"] += 1
        raise NoPrimaryError("no primary available: none of the configured controllers responded", "host1", 9991)

    monkeypatch.setattr(scheduler, "follow_redirects_any", fake_follow_redirects_any)

    with pytest.raises(NoPrimaryError):
        scheduler.record_status("host1:9991,host2:9992,host3:9993")

    assert calls["n"] == scheduler._MAX_NO_PRIMARY_ATTEMPTS
    assert len(sleep_calls) == scheduler._MAX_NO_PRIMARY_ATTEMPTS - 1


def test_record_status_standby_without_redirect_not_retried(mock_controller, wrap_reply, monkeypatch):
    """A standby rejection with NO populated redirect (the genuine "op is
    primary-only, no known primary to redirect to" case) is a hard,
    non-transient failure -- bug-1713's new retry logic must NOT retry it.
    Regression-locks E_XCE_STANDBY (-10014) being excluded from
    ``_TRANSIENT_STATUSES``. Uses the single-shot ``mock_controller``
    fixture deliberately: if this ever DID retry, the client's second
    connection attempt would stall against a listener with no more
    accepting thread, independently proving single-attempt behavior beyond
    just the ``sleep_calls`` assertion below.
    """
    mod = load_schema()
    sleep_calls: list[float] = []
    monkeypatch.setattr(scheduler.time, "sleep", lambda s: sleep_calls.append(s))

    def _handler(conn):
        wire.recv_frame(conn)
        reply = mod.RequestReply.new_message()
        reply.status = -10014  # E_XCE_STANDBY, errnox.h XCE_ERRNO_BASE+13
        reply.data.nodata = None
        wire.send_frame(conn, wrap_reply(reply))

    srv = mock_controller(_handler)

    with pytest.raises(SchedAdminError) as exc_info:
        scheduler.record_status(f"127.0.0.1:{srv.port}")

    assert "-10014" in exc_info.value.error
    assert sleep_calls == []


def test_record_status_standby_with_redirect_follows_redirect_not_retried(
    mock_controller, wrap_reply, monkeypatch
):
    """Mirrors the real controller's actual on-wire behavior for a standby
    hit that DOES know the primary (bug-1700/1701's fill_redirect_reply()
    call sends ``status=-XCE_ERR_STANDBY`` TOGETHER WITH a populated
    redirect) -- unlike ``test_record_start_follows_one_redirect_to_primary``'s
    ``status=0`` placeholder. Proves this fix does not disturb that
    already-fixed, live-verified path: the redirect is followed exactly as
    before, with zero involvement of the new retry logic (no sleep calls).
    """
    mod = load_schema()
    sleep_calls: list[float] = []
    monkeypatch.setattr(scheduler.time, "sleep", lambda s: sleep_calls.append(s))

    def _standby_handler(conn):
        wire.recv_frame(conn)
        reply = mod.RequestReply.new_message()
        reply.status = -10014  # E_XCE_STANDBY
        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-after-standby-redirect.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_status(f"127.0.0.1:{standby_srv.port}")

    assert result.file == "on-primary-after-standby-redirect.replay"
    assert sleep_calls == []


# ---------------------------------------------------------------------------
# bug-1715: a SECOND, differently-signaled transient family on top of
# bug-1713's numeric-status one above -- RequestReply.status == 0 (RPC
# succeeded) but SchedAdminReply{ok=false, error=<reason>} for one of the
# four confirmed-transient reason strings in
# ``scheduler._TRANSIENT_ADMIN_REASONS`` ("scheduler disconnected",
# "scheduler not connected", "request timeout", "server busy, try again").
# Reuses the exact same retry loop/budget/delay and the same
# ``sequential_mock_controller`` fixture bug-1713 introduced above --
# these replies also require a fresh reconnect per attempt.
# ---------------------------------------------------------------------------


@pytest.mark.parametrize(
    "transient_reason",
    sorted(scheduler._TRANSIENT_ADMIN_REASONS),
)
def test_record_status_retries_on_transient_admin_reason_then_succeeds(
    sequential_mock_controller, wrap_reply, monkeypatch, transient_reason
):
    """bug-1715's core proof: a mocked server returning
    ``ok=false, error=<transient_reason>`` twice, then succeeding on the
    3rd attempt, results in the CLIENT EVENTUALLY SUCCEEDING -- not an
    immediate raise. FAILS against the bug-1713-only code (which has no
    concept of ``_TRANSIENT_ADMIN_REASONS`` and raises SchedAdminError on
    the very first ok=false reply, regardless of reason string)."""
    mod = load_schema()
    sleep_calls: list[float] = []
    monkeypatch.setattr(scheduler.time, "sleep", lambda s: sleep_calls.append(s))

    def _transient_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 = transient_reason
        admin_reply.recording = False
        admin_reply.file = ""
        admin_reply.eventCount = 0
        admin_reply.startedAt = 0
        wire.send_frame(conn, wrap_reply(reply))

    def _success_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 = "settled-after-admin-retry.replay"
        admin_reply.eventCount = 0
        admin_reply.startedAt = 7000
        wire.send_frame(conn, wrap_reply(reply))

    srv = sequential_mock_controller([_transient_handler, _transient_handler, _success_handler])

    result = scheduler.record_status(f"127.0.0.1:{srv.port}")

    assert result.file == "settled-after-admin-retry.replay"
    assert srv.connection_count == 3
    assert sleep_calls == [scheduler._TRANSIENT_RETRY_DELAY_S] * 2


@pytest.mark.parametrize(
    "transient_reason",
    sorted(scheduler._TRANSIENT_ADMIN_REASONS),
)
def test_record_status_retry_exhaustion_on_transient_admin_reason_raises_distinguishable_error(
    sequential_mock_controller, wrap_reply, monkeypatch, transient_reason
):
    """A mocked server that ALWAYS returns the same transient admin reason
    (never recovers) must make the client eventually give up with a clear,
    distinguishable SchedAdminError after the retry budget is exhausted --
    not a silent hang or an infinite loop. ``time.sleep`` is monkeypatched
    so this test does not actually wait out the real retry delay."""
    mod = load_schema()
    sleep_calls: list[float] = []
    monkeypatch.setattr(scheduler.time, "sleep", lambda s: sleep_calls.append(s))

    def _transient_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 = transient_reason
        admin_reply.recording = False
        admin_reply.file = ""
        admin_reply.eventCount = 0
        admin_reply.startedAt = 0
        wire.send_frame(conn, wrap_reply(reply))

    # Exactly the retry budget's worth of handlers: connection_count is
    # asserted below to equal _MAX_TRANSIENT_ATTEMPTS exactly, proving the
    # client stops on its OWN bounded budget.
    srv = sequential_mock_controller([_transient_handler] * scheduler._MAX_TRANSIENT_ATTEMPTS)

    with pytest.raises(SchedAdminError) as exc_info:
        scheduler.record_status(f"127.0.0.1:{srv.port}")

    assert srv.connection_count == scheduler._MAX_TRANSIENT_ATTEMPTS
    assert len(sleep_calls) == scheduler._MAX_TRANSIENT_ATTEMPTS - 1
    # Distinguishable from an immediate, non-retried hard failure: the
    # message names both the original reason and that retries were
    # exhausted.
    assert transient_reason in exc_info.value.error
    assert "giving up" in exc_info.value.error
    assert f"{scheduler._MAX_TRANSIENT_ATTEMPTS} attempts" in exc_info.value.error


def test_record_stop_not_recording_never_retried(sequential_mock_controller, wrap_reply, monkeypatch):
    """bug-1715 non-regression: "not recording" (driver.c:426, rc==
    -EALREADY on an already-stopped recorder) is a genuine, PERMANENT
    domain rejection -- NOT in ``_TRANSIENT_ADMIN_REASONS`` -- so it must
    keep failing on the very first attempt, exactly like today's (pre-
    bug-1715) behavior, with zero retries/sleeps. Uses
    ``sequential_mock_controller`` (rather than the plain one-shot
    ``mock_controller``) specifically so ``connection_count`` can assert
    ``== 1`` explicitly, on top of the implicit proof that a real retry
    attempt against a server with only one handler queued would raise a
    connection error instead of ``SchedAdminError``.
    """
    mod = load_schema()
    sleep_calls: list[float] = []
    monkeypatch.setattr(scheduler.time, "sleep", lambda s: sleep_calls.append(s))

    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 = sequential_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 srv.connection_count == 1
    assert sleep_calls == []


# ---------------------------------------------------------------------------
# scheduler.py delegates --controller parsing to xceadmin.discovery.
# resolve_controllers (shared with push.py) -- see test_discovery.py for
# that function's own dedicated coverage; these two just confirm
# _send_admin_request's pipeline still reaches it for the single-host case.
# ---------------------------------------------------------------------------


def test_record_default_port_used_when_omitted():
    (host, port), = discovery.resolve_controllers("somehost")
    assert host == "somehost"
    assert port == 9991


def test_record_explicit_port_parsed():
    (host, port), = discovery.resolve_controllers("somehost:1234")
    assert host == "somehost"
    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)