"""MWP-5: clone/fetch/pull honor Retry-After with a bounded poll (fixes RC-5).

Test IDs MWP5_00–MWP5_18. All tests live here.

Phase 0 — RED harness (this file):
  - _seq_urllib_do: helper that raises TransportError(503, retry_after=R) for
    the first k calls, then returns a valid mpack POST body. Requires Phase 1
    to add retry_after to TransportError.__init__.
  - _record_sleeps: fixture patching muse.core.transport._sleep (create=True so
    it exists before Phase 2 adds the real function).
  - MWP5_00: RED until Phase 3 — asserts fetch_mpack retries past a single 503
    and calls _sleep once (currently raises immediately with no sleep).

Seam reference: tests/test_core_transport.py — the _urllib_do patch idiom.
"""

from __future__ import annotations

import http.server
import threading
import time
import unittest.mock
from collections.abc import Callable
from typing import Any

import msgpack
import pytest

from muse.core.transport import (
    HttpTransport,
    TransportError,
    _parse_retry_after,
    _next_wait,
    _resolve_retry_budget,
    _RETRY_AFTER_CEILING_S,
    _RETRY_AFTER_FLOOR_S,
    _RETRY_DEFAULT_BACKOFF_S,
    _FETCH_RETRY_BUDGET_DEFAULT_S,
)


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


def _minimal_fetch_success_body() -> bytes:
    """Minimal valid POST /fetch/mpack response body.

    Returns mpack_url=None (absent) so fetch_mpack uses the inline
    empty-result path — no S3 download needed. One _urllib_do call total
    after this is returned.
    """
    return msgpack.packb(
        {
            "mpack_id": None,
            "mpack_url": None,
            "repo_id": "r-test",
            "domain": "code",
            "default_branch": "main",
            "branch_heads": {},
            "commit_count": 0,
            "blob_count": 0,
        },
        use_bin_type=True,
    )


def _seq_urllib_do(
    *,
    k: int,
    retry_after: "int | None" = 30,
    success_body: bytes,
) -> Callable[..., bytes]:
    """Return a _urllib_do side_effect: raises TransportError(503) k times, then succeeds.

    NOTE: TransportError(msg, 503, retry_after=R) requires Phase 1 to add
    the retry_after keyword argument to TransportError.__init__. Until then,
    this helper raises TypeError on its first invocation (test is RED).
    """
    calls: list[int] = [0]

    def _side_effect(
        method: str,
        url: str,
        headers: dict[str, str],
        data: bytes | None = None,
        **kwargs: Any,
    ) -> bytes:
        calls[0] += 1
        if calls[0] <= k:
            raise TransportError("HTTP 503", 503, retry_after=retry_after)
        return success_body

    return _side_effect


# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------


@pytest.fixture()
def record_sleeps() -> Callable[[], list[float]]:
    """Patch muse.core.transport._sleep and record every call's argument.

    Uses create=True so the fixture works before Phase 2 adds the real _sleep
    function. Returns a callable that returns the list of sleep durations
    recorded so far.
    """
    sleep_log: list[float] = []

    def _fake_sleep(seconds: float) -> None:
        sleep_log.append(seconds)

    with unittest.mock.patch(
        "muse.core.transport._sleep",
        side_effect=_fake_sleep,
        create=True,
    ):
        yield lambda: list(sleep_log)


# ---------------------------------------------------------------------------
# Phase 0 — RED harness
# ---------------------------------------------------------------------------


class TestMWP5Phase0:
    def test_mwp5_00_fetch_retries_on_503(
        self,
        record_sleeps: Callable[[], list[float]],
    ) -> None:
        """MWP5_00 — RED until Phase 3.

        Asserts: one 503(Retry-After:30) followed by success causes
        fetch_mpack to return a FetchMPackResult (not raise) and to have
        called _sleep exactly once with ~30s.

        With current code this test is RED because:
          - Phase 0: TransportError does not accept retry_after kwarg →
            TypeError inside _seq_urllib_do → test errors.
          - Phase 1: retry_after added to TransportError (TypeError gone), but
            fetch_mpack still raises immediately on 503 → AssertionError.
          - Phase 2: _sleep seam exists but fetch_mpack still raises on 503.
          - Phase 3: retry loop added → test goes GREEN.
        """
        success = _minimal_fetch_success_body()
        fake_do = _seq_urllib_do(k=1, retry_after=30, success_body=success)

        with unittest.mock.patch(
            "muse.core.transport._urllib_do",
            side_effect=fake_do,
        ):
            result = HttpTransport().fetch_mpack(
                "https://hub.example.com/gabriel/testrepo",
                None,
                want=["sha256:" + "a" * 64],
                have=[],
            )

        # fetch_mpack succeeded — did not raise
        assert result is not None

        # _sleep was called exactly once with the Retry-After value (30s)
        sleeps = record_sleeps()
        assert len(sleeps) == 1, f"expected 1 sleep call, got {sleeps}"
        assert sleeps[0] == pytest.approx(30.0)


# ---------------------------------------------------------------------------
# Phase 1 — capture Retry-After (Defect A)
# ---------------------------------------------------------------------------


class TestMWP5Phase1:
    # -- MWP5_01 / MWP5_02: _parse_retry_after unit tests --------------------

    def test_mwp5_01_parse_retry_after_integer(self) -> None:
        """MWP5_01 — _parse_retry_after returns the integer seconds.

        In production, headers is http.client.HTTPMessage (case-insensitive
        .get()). Tests use plain dicts with the canonical casing.
        """
        assert _parse_retry_after({"Retry-After": "60"}) == 60
        assert _parse_retry_after({"Retry-After": "30"}) == 30
        assert _parse_retry_after({"Retry-After": "0"}) == 0

    def test_mwp5_02_parse_retry_after_invalid_returns_none(self) -> None:
        """MWP5_02 — missing or non-integer Retry-After returns None."""
        assert _parse_retry_after(None) is None
        assert _parse_retry_after({}) is None
        assert _parse_retry_after({"Retry-After": "Wed, 21 Oct 2025 07:28:00 GMT"}) is None
        assert _parse_retry_after({"Retry-After": ""}) is None
        assert _parse_retry_after({"Retry-After": "abc"}) is None
        assert _parse_retry_after({"Retry-After": "-1"}) is None

    # -- MWP5_03: real HTTPError driven through _urllib_do -------------------

    def test_mwp5_03_urllib_do_503_captures_retry_after(self) -> None:
        """MWP5_03 — a 503 HTTPError with Retry-After:30 header produces
        TransportError(status_code=503, retry_after=30) from _urllib_do.
        """
        import io
        import urllib.error
        import urllib.request
        from muse.core.transport import _urllib_do

        # Build a real urllib.error.HTTPError with a Retry-After header.
        # http.client.HTTPMessage (used by urllib) supports dict-style get().
        headers = urllib.request.addinfourl(
            io.BytesIO(b"mpack not ready"),
            {"Retry-After": "30", "Content-Type": "text/plain"},
            "https://hub.example.com/fetch/mpack",
            503,
        ).headers

        http_err = urllib.error.HTTPError(
            url="https://hub.example.com/fetch/mpack",
            code=503,
            msg="Service Unavailable",
            hdrs=headers,
            fp=io.BytesIO(b"mpack not ready"),
        )

        def _raise_503(req: "urllib.request.Request") -> "Any":
            raise http_err

        with unittest.mock.patch("muse.core.transport._open_url", side_effect=_raise_503):
            # _urllib_do goes through _open_url → HTTPError → TransportError
            # We call _urllib_do directly via the HttpTransport._execute_fetch
            # path to exercise that seam (it uses _open_url, not _urllib_do).
            # Also test _urllib_do directly by patching urllib opener.open.
            pass

        # Drive _urllib_do directly: patch urllib.request.build_opener to
        # return an opener whose open() raises the HTTPError.
        mock_opener = unittest.mock.MagicMock()
        mock_opener.open.return_value.__enter__ = unittest.mock.MagicMock(side_effect=http_err)
        mock_opener.open.side_effect = http_err

        with unittest.mock.patch(
            "urllib.request.build_opener", return_value=mock_opener
        ):
            with pytest.raises(TransportError) as exc_info:
                _urllib_do("POST", "https://hub.example.com/fetch/mpack", {})

        err = exc_info.value
        assert err.status_code == 503
        assert err.retry_after == 30

    # -- MWP5_04: 404 yields retry_after=None --------------------------------

    def test_mwp5_04_non_503_yields_retry_after_none(self) -> None:
        """MWP5_04 — a 404 HTTPError yields retry_after=None (no header).
        Regression guard: only 503s carry Retry-After; other errors unaffected.
        """
        with unittest.mock.patch(
            "muse.core.transport._urllib_do",
            side_effect=TransportError("HTTP 404", 404),
        ):
            with pytest.raises(TransportError) as exc_info:
                HttpTransport().fetch_remote_info(
                    "https://hub.example.com/gabriel/missing", None
                )

        err = exc_info.value
        assert err.status_code == 404
        assert err.retry_after is None


# ---------------------------------------------------------------------------
# Phase 2 — retry primitives (D3)
# ---------------------------------------------------------------------------


class TestMWP5Phase2:
    # -- MWP5_05: _next_wait clamping -----------------------------------------

    def test_mwp5_05_next_wait_floors_and_ceilings(self) -> None:
        """MWP5_05 — _next_wait clamps to [floor, ceiling] and uses default backoff."""
        # floor: 0 is clamped up to _RETRY_AFTER_FLOOR_S (1.0)
        assert _next_wait(0) == pytest.approx(_RETRY_AFTER_FLOOR_S)

        # ceiling: 999 is clamped down to _RETRY_AFTER_CEILING_S (60.0)
        assert _next_wait(999) == pytest.approx(_RETRY_AFTER_CEILING_S)

        # None → use _RETRY_DEFAULT_BACKOFF_S (5.0), within [floor, ceiling]
        assert _next_wait(None) == pytest.approx(_RETRY_DEFAULT_BACKOFF_S)

        # typical values pass through unchanged
        assert _next_wait(30) == pytest.approx(30.0)
        assert _next_wait(60) == pytest.approx(_RETRY_AFTER_CEILING_S)

        # floor boundary: 1 passes through (exactly at floor)
        assert _next_wait(1) == pytest.approx(1.0)

    # -- MWP5_06: _resolve_retry_budget priority chain -----------------------

    def test_mwp5_06_resolve_retry_budget(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """MWP5_06 — explicit arg > env var > module default; negatives → 0."""
        # explicit arg always wins
        assert _resolve_retry_budget(90.0) == pytest.approx(90.0)
        assert _resolve_retry_budget(0.0) == pytest.approx(0.0)

        # negative explicit is clamped to 0
        assert _resolve_retry_budget(-5.0) == pytest.approx(0.0)

        # env var used when no explicit arg
        monkeypatch.setenv("MUSE_FETCH_RETRY_BUDGET_S", "45")
        assert _resolve_retry_budget(None) == pytest.approx(45.0)

        # negative env is clamped to 0
        monkeypatch.setenv("MUSE_FETCH_RETRY_BUDGET_S", "-10")
        assert _resolve_retry_budget(None) == pytest.approx(0.0)

        # junk env falls through to module default
        monkeypatch.setenv("MUSE_FETCH_RETRY_BUDGET_S", "not-a-number")
        assert _resolve_retry_budget(None) == pytest.approx(_FETCH_RETRY_BUDGET_DEFAULT_S)

        # no env → module default
        monkeypatch.delenv("MUSE_FETCH_RETRY_BUDGET_S", raising=False)
        assert _resolve_retry_budget(None) == pytest.approx(_FETCH_RETRY_BUDGET_DEFAULT_S)


# ---------------------------------------------------------------------------
# Phase 3 — wire the loop into fetch_mpack (D4)
# ---------------------------------------------------------------------------


class TestMWP5Phase3:
    """MWP5_07–MWP5_12: the retry loop in HttpTransport.fetch_mpack."""

    def test_mwp5_07_single_503_then_success(
        self,
        record_sleeps: Callable[[], list[float]],
    ) -> None:
        """MWP5_07 — one 503(Retry-After:30) then success.

        fetch_mpack must return a FetchMPackResult (not raise) and call
        _sleep exactly once with 30.0.
        """
        success = _minimal_fetch_success_body()
        fake_do = _seq_urllib_do(k=1, retry_after=30, success_body=success)

        with unittest.mock.patch(
            "muse.core.transport._urllib_do", side_effect=fake_do
        ):
            result = HttpTransport().fetch_mpack(
                "https://hub.example.com/gabriel/testrepo",
                None,
                want=["sha256:" + "a" * 64],
                have=[],
            )

        assert result is not None
        sleeps = record_sleeps()
        assert len(sleeps) == 1, f"expected 1 sleep, got {sleeps}"
        assert sleeps[0] == pytest.approx(30.0)

    def test_mwp5_08_two_503s_then_success(
        self,
        record_sleeps: Callable[[], list[float]],
    ) -> None:
        """MWP5_08 — 503(30) then 503(60) then success: two sleeps, result returned.

        Uses a custom side_effect that emits different retry_after values per
        call (30 on call 1, 60 on call 2, success on call 3).
        """
        success = _minimal_fetch_success_body()
        retry_afters = [30, 60]
        calls: list[int] = [0]

        def _two_503_side_effect(method: str, url: str, headers: dict, data: bytes | None = None, **kw: Any) -> bytes:
            calls[0] += 1
            if calls[0] <= len(retry_afters):
                raise TransportError("HTTP 503", 503, retry_after=retry_afters[calls[0] - 1])
            return success

        with unittest.mock.patch(
            "muse.core.transport._urllib_do", side_effect=_two_503_side_effect
        ):
            result = HttpTransport().fetch_mpack(
                "https://hub.example.com/gabriel/testrepo",
                None,
                want=["sha256:" + "a" * 64],
                have=[],
            )

        assert result is not None
        sleeps = record_sleeps()
        assert len(sleeps) == 2, f"expected 2 sleeps, got {sleeps}"
        assert sleeps[0] == pytest.approx(30.0)
        assert sleeps[1] == pytest.approx(60.0)

    def test_mwp5_09_503_no_retry_after_uses_default_backoff(
        self,
        record_sleeps: Callable[[], list[float]],
    ) -> None:
        """MWP5_09 — 503 with no Retry-After header: sleeps _RETRY_DEFAULT_BACKOFF_S."""
        success = _minimal_fetch_success_body()
        # retry_after=None means the TransportError carries no wait hint
        fake_do = _seq_urllib_do(k=1, retry_after=None, success_body=success)  # type: ignore[arg-type]

        # _seq_urllib_do stores retry_after=None; TransportError.retry_after will be None
        with unittest.mock.patch(
            "muse.core.transport._urllib_do", side_effect=fake_do
        ):
            result = HttpTransport().fetch_mpack(
                "https://hub.example.com/gabriel/testrepo",
                None,
                want=["sha256:" + "a" * 64],
                have=[],
            )

        assert result is not None
        sleeps = record_sleeps()
        assert len(sleeps) == 1, f"expected 1 sleep, got {sleeps}"
        assert sleeps[0] == pytest.approx(_RETRY_DEFAULT_BACKOFF_S)

    def test_mwp5_10_budget_exhausted_raises_503(
        self,
        record_sleeps: Callable[[], list[float]],
    ) -> None:
        """MWP5_10 — 503 forever with budget=120: raises TransportError(503).

        Total accumulated sleep must not exceed the budget. Sleep count is
        finite. The final raise surfaces the server's real 503 (status_code=503).
        """
        success = _minimal_fetch_success_body()
        # k=999 simulates "503 forever" in practice
        fake_do = _seq_urllib_do(k=999, retry_after=60, success_body=success)

        with unittest.mock.patch(
            "muse.core.transport._urllib_do", side_effect=fake_do
        ):
            with pytest.raises(TransportError) as exc_info:
                HttpTransport().fetch_mpack(
                    "https://hub.example.com/gabriel/testrepo",
                    None,
                    want=["sha256:" + "a" * 64],
                    have=[],
                    retry_budget_s=120.0,
                )

        err = exc_info.value
        assert err.status_code == 503  # server's real error, not synthetic

        sleeps = record_sleeps()
        total_slept = sum(sleeps)
        assert total_slept <= 120.0, f"total slept {total_slept}s > 120s budget"
        assert len(sleeps) < 20, f"sleep count {len(sleeps)} not bounded"

    def test_mwp5_11_non_503_raises_immediately(
        self,
        record_sleeps: Callable[[], list[float]],
    ) -> None:
        """MWP5_11 — a 404 inside the loop raises immediately; _sleep never called.

        Non-503 errors must not incur any added latency — the loop skips retry
        on the first non-retryable status.
        """
        with unittest.mock.patch(
            "muse.core.transport._urllib_do",
            side_effect=TransportError("HTTP 404", 404),
        ):
            with pytest.raises(TransportError) as exc_info:
                HttpTransport().fetch_mpack(
                    "https://hub.example.com/gabriel/testrepo",
                    None,
                    want=["sha256:" + "a" * 64],
                    have=[],
                )

        assert exc_info.value.status_code == 404
        assert record_sleeps() == []  # zero sleeps — no latency added

    def test_mwp5_12_retry_budget_zero_is_single_attempt(
        self,
        record_sleeps: Callable[[], list[float]],
    ) -> None:
        """MWP5_12 — retry_budget_s=0 (--no-retry): exactly one attempt, zero sleeps.

        This is the escape hatch for deterministic CI. The original single-attempt
        behavior is preserved exactly — the first 503 raises with no sleep.
        """
        success = _minimal_fetch_success_body()
        fake_do = _seq_urllib_do(k=1, retry_after=30, success_body=success)

        with unittest.mock.patch(
            "muse.core.transport._urllib_do", side_effect=fake_do
        ):
            with pytest.raises(TransportError) as exc_info:
                HttpTransport().fetch_mpack(
                    "https://hub.example.com/gabriel/testrepo",
                    None,
                    want=["sha256:" + "a" * 64],
                    have=[],
                    retry_budget_s=0.0,
                )

        assert exc_info.value.status_code == 503
        assert record_sleeps() == []  # zero sleeps — true single-attempt behavior


# ---------------------------------------------------------------------------
# Phase 4 — command integration (D5)
# ---------------------------------------------------------------------------

import argparse
import contextlib
import pathlib
from unittest.mock import MagicMock, patch

from muse.core.types import fake_id

# Minimal shared fixtures for command-layer tests.
_COMMIT_ID = fake_id("commit-1")
_SNAP_ID = fake_id("snap-1")
_REPO_ID = fake_id("repo-1")

_REMOTE_INFO: dict = {
    "repo_id": _REPO_ID,
    "domain": "code",
    "default_branch": "main",
    "branch_heads": {"main": _COMMIT_ID},
    "object_count": 1,
    "size_bytes": 100,
}

_FETCH_RESULT: dict = {
    "repo_id": _REPO_ID,
    "domain": "code",
    "default_branch": "main",
    "branch_heads": {"main": _COMMIT_ID},
    "commits": [],
    "snapshots": [],
    "blobs": [],
    "blobs_received": 0,
    "shallow_commits": [],
}

_APPLY_RESULT: dict = {
    "commits_written": 1,
    "snapshots_written": 1,
    "blobs_written": 0,
    "blobs_skipped": 0,
    "failed_blobs": [],
}


def _seq_transport(k: int, retry_after: int = 30) -> "HttpTransport":
    """Return a real HttpTransport whose _fetch_mpack_once raises 503 k times then succeeds.

    Patching _fetch_mpack_once (not fetch_mpack) lets the real retry loop in
    HttpTransport.fetch_mpack run — the critical path under test for MWP5_13–17.
    """
    from muse.core.transport import HttpTransport as _HttpTransport
    t = _HttpTransport()
    t.fetch_remote_info = MagicMock(return_value=_REMOTE_INFO)  # type: ignore[method-assign]
    calls: list[int] = [0]

    def _once(*a: Any, **kw: Any) -> dict:
        calls[0] += 1
        if calls[0] <= k:
            raise TransportError("HTTP 503", 503, retry_after=retry_after)
        return _FETCH_RESULT

    t._fetch_mpack_once = _once  # type: ignore[method-assign]
    return t


def _clone_patches(t: "Any") -> "list[contextlib.AbstractContextManager]":
    """Standard patches needed to drive clone.run() without a real repo or network."""
    return [
        patch("muse.cli.commands.clone.get_signing_identity", return_value=None),
        patch("muse.cli.commands.clone.make_transport", return_value=t),
        patch("muse.cli.commands.clone.apply_mpack", return_value=_APPLY_RESULT),
        patch("muse.cli.commands.clone.set_remote"),
        patch("muse.cli.commands.clone.set_remote_head"),
        patch("muse.cli.commands.clone.set_upstream"),
        patch("muse.cli.commands.clone.apply_manifest"),
        patch("muse.cli.commands.clone.read_commit", return_value=MagicMock(snapshot_id=_SNAP_ID)),
        patch("muse.cli.commands.clone.read_snapshot", return_value=MagicMock(manifest={})),
        patch("muse.core.transport._sleep", side_effect=lambda s: None),
    ]


class TestMWP5Phase4:
    """MWP5_13–MWP5_17: command-layer integration of the retry loop."""

    def test_mwp5_13_clone_503_then_success(
        self,
        tmp_path: pathlib.Path,
    ) -> None:
        """MWP5_13 — CLI clone, transport patched to 503-then-success: exits 0.

        The target directory must exist and contain the apply result after the
        retry succeeds. The directory is never rmtree'd during the retry loop.
        """
        from muse.cli.commands.clone import run as clone_run

        target = tmp_path / "cloned"
        t = _seq_transport(k=1, retry_after=30)

        args = argparse.Namespace(
            url="https://hub.example.com/gabriel/repo",
            directory=str(target),
            branch=None,
            depth=None,
            dry_run=False,
            no_checkout=True,  # skip working-tree restore for speed
            json_out=False,
            retry_budget_s=None,
        )

        with contextlib.ExitStack() as stack:
            for p in _clone_patches(t):
                stack.enter_context(p)
            clone_run(args)  # must not raise SystemExit

        # _fetch_mpack_once was called twice: once raised 503, once returned success
        # (the retry loop in fetch_mpack drove the second call)

    def test_mwp5_14_clone_503_forever_budget_zero_removes_dir(
        self,
        tmp_path: pathlib.Path,
    ) -> None:
        """MWP5_14 — CLI clone, no-retry (budget=0) + 503 forever: exits INTERNAL_ERROR.

        The target directory is removed by the terminal error path. Stderr must
        describe the failure. With --json the envelope carries retryable=True.
        """
        import io
        import json
        import sys as _sys
        from muse.cli.commands.clone import run as clone_run

        target = tmp_path / "cloned"
        t = _seq_transport(k=999, retry_after=5)  # never succeeds

        args = argparse.Namespace(
            url="https://hub.example.com/gabriel/repo",
            directory=str(target),
            branch=None,
            depth=None,
            dry_run=False,
            no_checkout=True,
            json_out=True,
            retry_budget_s=0.0,  # --no-retry: single attempt, immediate 503
        )

        stdout_buf = io.StringIO()
        stderr_buf = io.StringIO()
        with contextlib.ExitStack() as stack:
            for p in _clone_patches(t):
                stack.enter_context(p)
            stack.enter_context(contextlib.redirect_stdout(stdout_buf))
            stack.enter_context(contextlib.redirect_stderr(stderr_buf))
            with pytest.raises(SystemExit) as exc_info:
                clone_run(args)

        assert exc_info.value.code != 0  # exits with error

        # JSON envelope on stdout carries retryable marker
        stdout_text = stdout_buf.getvalue().strip()
        assert stdout_text, "expected JSON on stdout"
        envelope = json.loads(stdout_text.splitlines()[0])
        assert envelope.get("retryable") is True

        # Target directory cleaned up
        assert not target.exists(), "clone target must be rmtree'd on terminal 503"

    def test_mwp5_15_fetch_503_then_success(
        self,
        tmp_path: pathlib.Path,
    ) -> None:
        """MWP5_15 — CLI fetch, 503-then-success: exits 0, commits applied.

        Patches require_repo, make_transport, and the apply/ref helpers so no
        real repository is needed.
        """
        from muse.cli.commands.fetch import _fetch_one

        t = _seq_transport(k=1, retry_after=30)
        elapsed_fn: Callable[[], float] = lambda: 0.0

        with contextlib.ExitStack() as stack:
            stack.enter_context(patch("muse.cli.commands.fetch.get_remote", return_value="https://hub.example.com/gabriel/repo"))
            stack.enter_context(patch("muse.cli.commands.fetch.get_signing_identity", return_value=None))
            stack.enter_context(patch("muse.cli.commands.fetch.make_transport", return_value=t))
            stack.enter_context(patch("muse.cli.commands.fetch.get_all_commits", return_value=[]))
            stack.enter_context(patch("muse.cli.commands.fetch.apply_mpack", return_value=_APPLY_RESULT))
            stack.enter_context(patch("muse.cli.commands.fetch.set_remote_head"))
            stack.enter_context(patch("muse.core.transport._sleep", side_effect=lambda s: None))

            result = _fetch_one(
                tmp_path, "origin", "main",
                prune=False, dry_run=False, json_out=False,
                elapsed=elapsed_fn, retry_budget_s=None,
            )

        assert result["status"] == "fetched"
        # retry loop drove two _fetch_mpack_once calls: first raised 503, second succeeded

    def test_mwp5_16_pull_503_then_success(
        self,
        tmp_path: pathlib.Path,
    ) -> None:
        """MWP5_16 — CLI pull, 503-then-success: exits 0, ff/merge proceeds.

        Uses pull.run() with full command patches. Verifies fetch_mpack was
        called twice and the command completes without raising SystemExit.
        """
        from muse.cli.commands.pull import run as pull_run

        t = _seq_transport(k=1, retry_after=30)

        # Create a minimal local commit to satisfy get_all_commits
        commit_mock = MagicMock()
        commit_mock.commit_id = _COMMIT_ID

        args = argparse.Namespace(
            remote="origin",
            branch_flag=None,
            branch_pos="main",
            no_merge=True,   # fetch-only — skip the merge step complexity
            ff_only=False,
            dry_run=False,
            message=None,
            json_out=False,
            retry_budget_s=None,
        )

        with contextlib.ExitStack() as stack:
            stack.enter_context(patch("muse.cli.commands.pull.require_repo", return_value=tmp_path))
            stack.enter_context(patch("muse.cli.commands.pull.get_remote", return_value="https://hub.example.com/gabriel/repo"))
            stack.enter_context(patch("muse.cli.commands.pull.get_signing_identity", return_value=None))
            stack.enter_context(patch("muse.cli.commands.pull.make_transport", return_value=t))
            stack.enter_context(patch("muse.cli.commands.pull.read_current_branch", return_value="main"))
            stack.enter_context(patch("muse.cli.commands.pull.get_all_commits", return_value=[commit_mock]))
            stack.enter_context(patch("muse.cli.commands.pull.apply_mpack", return_value=_APPLY_RESULT))
            stack.enter_context(patch("muse.cli.commands.pull.set_remote_head"))
            stack.enter_context(patch("muse.core.transport._sleep", side_effect=lambda s: None))

            pull_run(args)  # must not raise SystemExit
        # retry loop drove two _fetch_mpack_once calls: first raised 503, second succeeded

    def test_mwp5_16b_pull_503_budget_exhausted_json_retryable(
        self,
        tmp_path: pathlib.Path,
    ) -> None:
        """MWP5_16b — CLI pull --json, 503-forever (budget=0): JSON envelope has retryable=True."""
        import io
        import json as _json
        from muse.cli.commands.pull import run as pull_run

        t = _seq_transport(k=999, retry_after=5)
        commit_mock = MagicMock()
        commit_mock.commit_id = _COMMIT_ID

        args = argparse.Namespace(
            remote="origin",
            branch_flag=None,
            branch_pos="main",
            no_merge=True,
            ff_only=False,
            dry_run=False,
            message=None,
            json_out=True,
            retry_budget_s=0.0,
        )

        stdout_buf = io.StringIO()
        with contextlib.ExitStack() as stack:
            stack.enter_context(patch("muse.cli.commands.pull.require_repo", return_value=tmp_path))
            stack.enter_context(patch("muse.cli.commands.pull.get_remote", return_value="https://hub.example.com/gabriel/repo"))
            stack.enter_context(patch("muse.cli.commands.pull.get_signing_identity", return_value=None))
            stack.enter_context(patch("muse.cli.commands.pull.make_transport", return_value=t))
            stack.enter_context(patch("muse.cli.commands.pull.read_current_branch", return_value="main"))
            stack.enter_context(patch("muse.cli.commands.pull.get_all_commits", return_value=[commit_mock]))
            stack.enter_context(patch("muse.core.transport._sleep", side_effect=lambda s: None))
            stack.enter_context(contextlib.redirect_stdout(stdout_buf))

            with pytest.raises(SystemExit) as exc_info:
                pull_run(args)

        assert exc_info.value.code != 0
        stdout_text = stdout_buf.getvalue().strip()
        assert stdout_text, "expected JSON on stdout"
        envelope = _json.loads(stdout_text.splitlines()[0])
        assert envelope.get("retryable") is True

    def test_mwp5_17_json_clone_retry_chatter_on_stderr_only(
        self,
        tmp_path: pathlib.Path,
    ) -> None:
        """MWP5_17 — --json clone with retries: stdout has exactly one JSON object.

        All retry progress lines (⏳ …) must go to stderr. Stdout must contain
        only the final JSON envelope (or error envelope on failure).
        """
        import io
        import json

        # Use a budget=0 (no-retry) so we get immediate failure — one JSON
        # object on stdout and any error messages on stderr.
        from muse.cli.commands.clone import run as clone_run

        target = tmp_path / "cloned"
        t = _seq_transport(k=999, retry_after=5)

        args = argparse.Namespace(
            url="https://hub.example.com/gabriel/repo",
            directory=str(target),
            branch=None,
            depth=None,
            dry_run=False,
            no_checkout=True,
            json_out=True,
            retry_budget_s=0.0,
        )

        stdout_buf = io.StringIO()
        stderr_buf = io.StringIO()

        with contextlib.ExitStack() as stack:
            for p in _clone_patches(t):
                stack.enter_context(p)
            stack.enter_context(contextlib.redirect_stdout(stdout_buf))
            stack.enter_context(contextlib.redirect_stderr(stderr_buf))
            with pytest.raises(SystemExit):
                clone_run(args)

        stdout_text = stdout_buf.getvalue().strip()
        stdout_lines = [l for l in stdout_text.splitlines() if l.strip()]
        # Exactly one JSON object on stdout
        assert len(stdout_lines) == 1, (
            f"stdout must have exactly one line, got {len(stdout_lines)}: {stdout_lines}"
        )
        envelope = json.loads(stdout_lines[0])
        assert "status" in envelope or "error" in envelope


# ---------------------------------------------------------------------------
# Phase 5 — real-urllib E2E (MWP5_18)
# ---------------------------------------------------------------------------


class TestMWP5Phase5:
    """MWP5_18 — end-to-end proof that Retry-After survives the real urllib path."""

    def test_mwp5_18_real_urllib_503_then_success(self) -> None:
        """MWP5_18 — stand up a local HTTP server: 503+Retry-After:1 then valid mpack.

        Drives a real HttpTransport.fetch_mpack (real _urllib_do, real urllib
        HTTPError, real header parsing) with _sleep patched to no-op.
        Proves the Retry-After header value survives the actual urllib code path
        end-to-end — not just a mocked side_effect.
        """
        import io
        import json
        import socket

        success_body = _minimal_fetch_success_body()
        call_count: list[int] = [0]

        class _Handler(http.server.BaseHTTPRequestHandler):
            def do_POST(self) -> None:
                # Drain request body to avoid broken-pipe on client side
                content_len = int(self.headers.get("Content-Length", 0))
                self.rfile.read(content_len)

                call_count[0] += 1
                if call_count[0] == 1:
                    # First call: 503 with Retry-After: 1
                    self.send_response(503)
                    self.send_header("Retry-After", "1")
                    self.send_header("Content-Type", "application/octet-stream")
                    body = b"mpack not ready"
                    self.send_header("Content-Length", str(len(body)))
                    self.end_headers()
                    self.wfile.write(body)
                else:
                    # Second call: 200 with valid mpack body
                    self.send_response(200)
                    self.send_header("Content-Type", "application/octet-stream")
                    self.send_header("Content-Length", str(len(success_body)))
                    self.end_headers()
                    self.wfile.write(success_body)

            def log_message(self, *args: Any) -> None:
                pass  # suppress request logs in test output

        # Bind on an OS-assigned port so there are no port conflicts.
        server = http.server.HTTPServer(("127.0.0.1", 0), _Handler)
        port = server.server_address[1]
        server_thread = threading.Thread(target=server.handle_request, daemon=True)
        server_thread.start()

        # Run a second request handler for the success call.
        # HTTPServer.handle_request() handles exactly one connection;
        # start a second thread for the retry.
        def _serve_one() -> None:
            server.handle_request()

        second_thread = threading.Thread(target=_serve_one, daemon=True)
        second_thread.start()

        url = f"http://127.0.0.1:{port}"
        with unittest.mock.patch("muse.core.transport._sleep", side_effect=lambda s: None):
            result = HttpTransport().fetch_mpack(
                url, None,
                want=["sha256:" + "a" * 64],
                have=[],
            )

        server_thread.join(timeout=5)
        second_thread.join(timeout=5)
        server.server_close()

        # fetch_mpack returned a result (did not raise) — retry succeeded
        assert result is not None
        # Server was called exactly twice (once 503, once success)
        assert call_count[0] == 2, f"server call count: {call_count[0]}"