"""Tests for the endpoint chat app.

Run: pytest -q (the required environment is filled in below)
"""

import json
import os

import pytest
import requests

# Assigned outright so real values exported in the shell cannot leak in.
os.environ.update(
    ENDPOINT_URL="http://endpoint.test/",
    QWEN_ENDPOINT_URL="http://qwen.endpoint.test/",
    DEEPSEEK_ENDPOINT_URL="http://deepseek.endpoint.test/",
    HF_TOKEN="test-token",
    QWEN_HF_TOKEN="qwen-org-token",
)
os.environ.pop("DEEPSEEK_HF_TOKEN", None)
os.environ.setdefault("LOGIN_PASSWORD", "s3cret")

import app


def drain(generator):
    """Consume a streaming handler, returning (all yields, final yield)."""
    frames = list(generator)
    return frames, frames[-1]


def histories(frames):
    return [frame[0] for frame in frames]


def textbox(frame):
    """The value the textbox update carries."""
    update = frame[1]
    return update["value"] if isinstance(update, dict) else update


def locked(frame):
    """True when the frame locks the input controls (a request is in flight)."""
    msg_update, send_update, stop_update = frame[1:]
    return (
        msg_update["interactive"] is False
        and send_update["interactive"] is False
        and stop_update["interactive"] is True
    )


def drain_failing(generator, expected):
    """Consume a streaming handler that fails partway; returns frames seen."""
    frames = []

    with pytest.raises(expected):
        for frame in generator:
            frames.append(frame)  # noqa: PERF402 - loop can raise mid-iteration

    return frames


def fake_stream(*chunks):
    def stream(messages, model_name, turn="-"):
        yield from chunks

    return stream


def test_build_api_messages_keeps_order_and_appends_new_turn():
    history = [
        {"role": "user", "content": "hi"},
        {"role": "assistant", "content": "hello"},
    ]

    assert app.build_api_messages(history, "how are you?") == [
        {"role": "user", "content": "hi"},
        {"role": "assistant", "content": "hello"},
        {"role": "user", "content": "how are you?"},
    ]


def test_build_api_messages_flattens_gradio_typed_content():
    """Gradio hands handlers typed content parts, not plain strings."""
    import gradio as gr

    chatbot = gr.Chatbot()
    history = chatbot.preprocess(
        chatbot.postprocess(
            [
                {"role": "user", "content": "hi"},
                {"role": "assistant", "content": "hello"},
            ]
        )
    )

    assert history[0]["content"] != "hi", "test no longer exercises typed parts"
    assert app.build_api_messages(history, "next") == [
        {"role": "user", "content": "hi"},
        {"role": "assistant", "content": "hello"},
        {"role": "user", "content": "next"},
    ]


def test_build_api_messages_drops_non_text_and_malformed_entries():
    history = [
        {"role": "user", "content": [{"type": "file", "file": {"path": "a.png"}}]},
        {"role": "system", "content": "ignored"},
        {"role": "user"},
        None,
        {"role": "assistant", "content": "kept"},
    ]

    assert app.build_api_messages(history, "next") == [
        {"role": "assistant", "content": "kept"},
        {"role": "user", "content": "next"},
    ]


def test_message_text_handles_mixed_and_unknown_content():
    assert app.message_text("plain") == "plain"
    assert app.message_text([{"type": "text", "text": "a"}, {"type": "file"}]) == "a"
    assert (
        app.message_text([{"type": "text", "text": "a"}, {"type": "text", "text": "b"}])
        == "ab"
    )
    assert app.message_text([{"type": "text", "text": None}]) == ""
    assert app.message_text(None) == ""
    assert app.message_text(42) == ""


def test_respond_streams_partial_replies_then_final(monkeypatch):
    monkeypatch.setattr(app, "stream_completion", fake_stream("Hel", "lo ", "there"))

    frames, final = drain(app.respond("ping", [], app.DEFAULT_MODEL))
    history = final[0]

    # Assistant text grows monotonically across frames.
    assert [h[-1]["content"] for h in histories(frames)] == [
        app.WAITING_TEXT,
        "Hel",
        "Hello ",
        "Hello there",
        "Hello there",
    ]
    assert history == [
        {"role": "user", "content": "ping"},
        {"role": "assistant", "content": "Hello there"},
    ]
    assert textbox(final) == ""


def test_respond_shows_pending_spinner_until_first_token(monkeypatch):
    monkeypatch.setattr(app, "stream_completion", fake_stream("first token"))

    frames, final = drain(app.respond("ping", [], app.DEFAULT_MODEL))

    # Frame 1: user turn plus a pending placeholder that renders as a spinner.
    placeholder = frames[0][0][-1]
    assert placeholder["metadata"]["status"] == "pending"
    assert placeholder["metadata"]["title"]

    # Regression: the Chatbot with reasoning_tags silently dropped the empty
    # placeholder, so a sleeping endpoint showed no feedback at all.
    import gradio as gr

    chatbot = next(b for b in app.demo.blocks.values() if isinstance(b, gr.Chatbot))
    rendered = chatbot.postprocess(frames[0][0]).root
    assert [m.metadata.get("status") if m.metadata else None for m in rendered] == [
        None,
        "pending",
    ]

    # The spinner is gone the moment real text exists, and never comes back.
    for history in histories(frames[1:]):
        assert "metadata" not in history[-1], "spinner outlived the first token"

    assert "metadata" not in final[0][-1]


def test_respond_locks_controls_only_while_in_flight(monkeypatch):
    monkeypatch.setattr(app, "stream_completion", fake_stream("a", "b"))

    frames, final = drain(app.respond("ping", [], app.DEFAULT_MODEL))

    assert [locked(f) for f in frames] == [True, True, True, False]
    assert locked(final) is False


def test_respond_unlocks_controls_when_message_is_blank(monkeypatch):
    monkeypatch.setattr(app, "stream_completion", fake_stream("unused"))

    _frames, final = drain(app.respond("", [], app.DEFAULT_MODEL))

    assert locked(final) is False


def test_respond_rejects_an_empty_reply(monkeypatch):
    import gradio as gr

    monkeypatch.setattr(app, "stream_completion", fake_stream())

    frames = drain_failing(app.respond("ping", [], app.DEFAULT_MODEL), gr.Error)

    # No stuck spinner, no empty bubble, and the controls are released.
    assert frames[-1][0] == []
    assert locked(frames[-1]) is False


def test_respond_history_survives_gradio_format_check(monkeypatch):
    """Regression: tuple history raised 'Data incompatible with messages format'."""
    import gradio as gr

    monkeypatch.setattr(app, "stream_completion", fake_stream("pong"))

    frames, _ = drain(app.respond("ping", [], app.DEFAULT_MODEL))

    for history in histories(frames):
        gr.Chatbot()._check_format(history)


def test_respond_sends_history_without_the_placeholder_turn(monkeypatch):
    sent = {}

    def stream(messages, model_name, turn="-"):
        sent["messages"] = messages
        yield "ok"

    monkeypatch.setattr(app, "stream_completion", stream)

    drain(app.respond("second", [{"role": "user", "content": "first"}], None))

    assert sent["messages"] == [
        {"role": "user", "content": "first"},
        {"role": "user", "content": "second"},
    ]


def test_respond_ignores_blank_message(monkeypatch):
    def fail(*args, **kwargs):
        raise AssertionError("endpoint should not be called for a blank message")

    monkeypatch.setattr(app, "stream_completion", fail)

    _frames, final = drain(
        app.respond("   ", [{"role": "user", "content": "hi"}], None)
    )

    assert final[0] == [{"role": "user", "content": "hi"}]


def test_respond_does_not_mutate_caller_history(monkeypatch):
    monkeypatch.setattr(app, "stream_completion", fake_stream("pong"))
    history = []

    drain(app.respond("ping", history, app.DEFAULT_MODEL))

    assert history == []


@pytest.mark.parametrize(
    "error",
    [
        requests.exceptions.Timeout(),
        requests.exceptions.ConnectionError("refused"),
        ValueError("not json"),
    ],
)
def test_respond_surfaces_immediate_failures_and_restores_input(monkeypatch, error):
    import gradio as gr

    def stream(messages, model_name, turn="-"):
        raise error
        yield  # pragma: no cover - makes this a generator

    monkeypatch.setattr(app, "stream_completion", stream)

    frames = drain_failing(app.respond("ping", [], app.DEFAULT_MODEL), gr.Error)

    # Nothing streamed, so the whole failed turn is rolled back and the user's
    # text is handed back to the input box rather than duplicated in the chat.
    assert frames[-1][0] == []
    assert textbox(frames[-1]) == "ping"
    assert locked(frames[-1]) is False


def test_respond_keeps_partial_reply_when_stream_breaks(monkeypatch):
    import gradio as gr

    def stream(messages, model_name, turn="-"):
        yield "half a "
        raise requests.exceptions.ChunkedEncodingError("dropped")

    monkeypatch.setattr(app, "stream_completion", stream)

    frames = drain_failing(app.respond("ping", [], app.DEFAULT_MODEL), gr.Error)

    assert frames[-1][0] == [
        {"role": "user", "content": "ping"},
        {"role": "assistant", "content": "half a "},
    ]
    # A kept partial reply means the box is not refilled with the same text.
    assert textbox(frames[-1]) == ""
    assert locked(frames[-1]) is False


def test_respond_reraises_unexpected_errors(monkeypatch):
    def stream(messages, model_name, turn="-"):
        raise RuntimeError("boom")
        yield  # pragma: no cover

    monkeypatch.setattr(app, "stream_completion", stream)

    with pytest.raises(RuntimeError, match="boom"):
        drain(app.respond("ping", [], app.DEFAULT_MODEL))


def test_verify_login_rejects_wrong_and_empty_password():
    assert app.verify_login("user", "s3cret") is True
    assert app.verify_login("user", "wrong") is False
    assert app.verify_login("user", "") is False
    assert app.verify_login("user", None) is False


def test_missing_required_env_fails_with_named_message(monkeypatch):
    monkeypatch.delenv("ENDPOINT_URL", raising=False)

    with pytest.raises(RuntimeError, match="ENDPOINT_URL"):
        app._required_env("ENDPOINT_URL")


class FakeStreamResponse:
    """Stand-in for the streaming `requests` response."""

    def __init__(self, lines, status_code=200):
        self._lines = lines
        self.status_code = status_code
        self.closed = False

    def __enter__(self):
        return self

    def __exit__(self, *exc_info):
        self.closed = True

    def raise_for_status(self):
        pass

    def iter_lines(self, chunk_size=512, decode_unicode=False):
        self.iter_kwargs = {"chunk_size": chunk_size, "decode_unicode": decode_unicode}
        yield from self._lines


def sse(*payloads):
    return [f"data: {p}" for p in payloads]


def chunk(content):
    return json.dumps({"choices": [{"delta": {"content": content}}]})


def test_each_model_posts_to_its_own_endpoint(monkeypatch):
    captured = {}

    def fake_post(url, **kwargs):
        captured.update(url=url, **kwargs)
        return FakeStreamResponse(sse(chunk("hi"), "[DONE]"))

    monkeypatch.setattr(app.requests, "post", fake_post)

    expected = {
        app.GEMMA_MODEL: "http://endpoint.test",
        app.QWEN_MODEL: "http://qwen.endpoint.test",
        app.DEEPSEEK_MODEL: "http://deepseek.endpoint.test",
    }

    for model, url in expected.items():
        list(app.stream_completion([{"role": "user", "content": "hi"}], model))
        assert captured["url"] == f"{url}/v1/chat/completions"
        assert captured["json"]["model"] == model

    assert app.AVAILABLE_MODELS == [
        app.GEMMA_MODEL,
        app.QWEN_MODEL,
        app.DEEPSEEK_MODEL,
    ]


def test_each_model_sends_its_own_token(monkeypatch):
    """Endpoints in different orgs need tokens with access to that org."""
    sent = {}

    def fake_post(url, **kwargs):
        sent[url] = kwargs["headers"]["Authorization"]
        return FakeStreamResponse(sse(chunk("hi"), "[DONE]"))

    monkeypatch.setattr(app.requests, "post", fake_post)

    for model in app.AVAILABLE_MODELS:
        list(app.stream_completion([], model))

    assert sent == {
        "http://endpoint.test/v1/chat/completions": "Bearer test-token",
        "http://qwen.endpoint.test/v1/chat/completions": "Bearer qwen-org-token",
        # No DEEPSEEK_HF_TOKEN, so it falls back to HF_TOKEN.
        "http://deepseek.endpoint.test/v1/chat/completions": "Bearer test-token",
    }
    assert app.MODELS[app.QWEN_MODEL].token_env == "QWEN_HF_TOKEN"
    assert app.MODELS[app.DEEPSEEK_MODEL].token_env == "HF_TOKEN"


@pytest.mark.parametrize("status", [401, 403])
def test_rejected_token_names_the_secret_to_fix(monkeypatch, status):
    import gradio as gr

    monkeypatch.setattr(
        app.requests, "post", lambda url, **kw: FakeStreamResponse([], status)
    )

    with pytest.raises(gr.Error, match="QWEN_HF_TOKEN") as error:
        list(app.stream_completion([], app.QWEN_MODEL))

    assert "qwen-org-token" not in str(error.value)


def test_stream_completion_parses_sse_chunks(monkeypatch):
    lines = [
        ": keep-alive",
        "",
        *sse(chunk("Hel"), chunk("lo")),
        "",
        # Final chunk carries a finish_reason and no content.
        json.dumps({"choices": [{"delta": {}, "finish_reason": "stop"}]}).join(
            ["data: ", ""]
        ),
        "data: [DONE]",
        "data: {bad json that must never be reached}",
    ]
    response = FakeStreamResponse(lines)
    captured = {}

    def fake_post(url, **kwargs):
        captured.update(url=url, **kwargs)
        return response

    monkeypatch.setattr(app.requests, "post", fake_post)

    assert list(app.stream_completion([{"role": "user", "content": "hi"}], "m")) == [
        "Hel",
        "lo",
    ]
    assert captured["url"] == "http://endpoint.test/v1/chat/completions"
    assert captured["json"]["stream"] is True
    assert captured["stream"] is True
    assert captured["headers"]["Authorization"] == "Bearer test-token"
    assert response.closed


def reasoning_chunk(text):
    return json.dumps({"choices": [{"delta": {"reasoning_content": text}}]})


def test_stream_completion_wraps_reasoning_in_think_tags(monkeypatch):
    """llama.cpp streams thinking as reasoning_content; the Chatbot needs tags."""
    lines = sse(reasoning_chunk("7 x "), reasoning_chunk("143"), chunk("No."), "[DONE]")
    monkeypatch.setattr(
        app.requests, "post", lambda url, **kwargs: FakeStreamResponse(lines)
    )

    assert "".join(app.stream_completion([], "m")) == "<think>7 x 143</think>No."


def test_stream_completion_closes_reasoning_cut_off_mid_thought(monkeypatch):
    lines = sse(reasoning_chunk("still thinking"), "[DONE]")
    monkeypatch.setattr(
        app.requests, "post", lambda url, **kwargs: FakeStreamResponse(lines)
    )

    assert "".join(app.stream_completion([], "m")) == "<think>still thinking</think>"


def test_reasoning_is_not_sent_back_to_the_model():
    """The Chatbot splits reasoning into a titled thought, which is dropped."""
    import gradio as gr

    chatbot = gr.Chatbot(reasoning_tags=[(app.THINK_OPEN, app.THINK_CLOSE)])
    history = chatbot.preprocess(
        chatbot.postprocess(
            [
                {"role": "user", "content": "hi"},
                {"role": "assistant", "content": "<think>hmm</think>hello"},
            ]
        )
    )

    assert app.build_api_messages(history, "next") == [
        {"role": "user", "content": "hi"},
        {"role": "assistant", "content": "hello"},
        {"role": "user", "content": "next"},
    ]


def test_stop_replaces_send_while_a_reply_streams():
    _msg, send, stop = app.busy_controls()
    assert send["visible"] is False and stop["visible"] is True

    _msg, send, stop = app.idle_controls()
    assert send["visible"] is True and stop["visible"] is False


def test_new_chat_empties_history_and_unlocks():
    history, *controls = app.new_chat()

    assert history == []
    assert controls[0]["interactive"] is True


def test_every_model_has_a_display_label():
    assert set(app.MODEL_LABELS) == set(app.AVAILABLE_MODELS)


def test_stream_completion_tolerates_chunks_without_choices(monkeypatch):
    lines = sse(json.dumps({"choices": []}), chunk("ok"), "[DONE]")
    monkeypatch.setattr(
        app.requests, "post", lambda url, **kwargs: FakeStreamResponse(lines)
    )

    assert list(app.stream_completion([], "m")) == ["ok"]


def test_stream_completion_raises_on_malformed_chunk(monkeypatch):
    lines = sse(chunk("ok"), "{not json")
    monkeypatch.setattr(
        app.requests, "post", lambda url, **kwargs: FakeStreamResponse(lines)
    )

    with pytest.raises(ValueError):
        list(app.stream_completion([], "m"))


def test_module_exposes_demo_for_spaces_hot_reload():
    """Spaces' reloader does getattr(__main__, "demo") after every reload."""
    import gradio as gr

    assert isinstance(app.demo, gr.Blocks)


def test_cancel_stream_clears_spinner_and_unlocks(monkeypatch):
    """Stop cancels the generator mid-flight, so cleanup runs in its own event."""
    monkeypatch.setattr(app, "stream_completion", fake_stream("half"))

    frames, _final = drain(app.respond("ping", [], app.DEFAULT_MODEL))
    mid_flight = frames[0][0]  # still showing the pending spinner

    history, *controls = app.cancel_stream(mid_flight)

    assert history == [{"role": "user", "content": "ping"}]
    assert all(update["interactive"] is not False for update in controls[:2])


def test_normalize_history_drops_thought_placeholders():
    history = [
        {"role": "user", "content": "hi"},
        {"role": "assistant", "content": "", "metadata": app.WAITING_METADATA},
        {"role": "assistant", "content": "kept", "metadata": {}},
    ]

    assert app.normalize_history(history) == [
        {"role": "user", "content": "hi"},
        {"role": "assistant", "content": "kept"},
    ]


def test_stream_completion_reads_incrementally(monkeypatch):
    """Regression: the default 512-byte reads deliver the whole reply at once."""
    response = FakeStreamResponse(sse(chunk("hi"), "[DONE]"))
    monkeypatch.setattr(app.requests, "post", lambda url, **kwargs: response)

    list(app.stream_completion([], "m"))

    assert response.iter_kwargs["chunk_size"] == 1
    assert response.iter_kwargs["decode_unicode"] is True


def test_successful_turn_is_logged(monkeypatch, caplog):
    monkeypatch.setattr(app, "stream_completion", fake_stream("pong"))

    with caplog.at_level("INFO", logger="gemma-endpoint"):
        drain(app.respond("ping", [], app.DEFAULT_MODEL))

    messages = [r.getMessage() for r in caplog.records]
    assert any("start model=" in m for m in messages)
    assert any("complete reply_chars=4" in m for m in messages)


def test_failed_turn_is_logged_at_error(monkeypatch, caplog):
    import gradio as gr

    def stream(messages, model_name, turn="-"):
        raise requests.exceptions.ConnectionError("refused")
        yield  # pragma: no cover

    monkeypatch.setattr(app, "stream_completion", stream)

    with caplog.at_level("INFO", logger="gemma-endpoint"):
        drain_failing(app.respond("ping", [], app.DEFAULT_MODEL), gr.Error)

    errors = [r.getMessage() for r in caplog.records if r.levelname == "ERROR"]
    assert any("failed after" in m and "refused" in m for m in errors)


def test_unexpected_error_logs_a_traceback(monkeypatch, caplog):
    def stream(messages, model_name, turn="-"):
        raise RuntimeError("boom")
        yield  # pragma: no cover

    monkeypatch.setattr(app, "stream_completion", stream)

    with caplog.at_level("INFO", logger="gemma-endpoint"), pytest.raises(RuntimeError):
        drain(app.respond("ping", [], app.DEFAULT_MODEL))

    assert any(r.exc_info for r in caplog.records), "no traceback was logged"


def test_turn_ids_correlate_lines_of_one_turn(monkeypatch, caplog):
    def stream(messages, model_name, turn="-"):
        app.logger.info("turn=%s inner", turn)
        yield "ok"

    monkeypatch.setattr(app, "stream_completion", stream)

    with caplog.at_level("INFO", logger="gemma-endpoint"):
        drain(app.respond("one", [], app.DEFAULT_MODEL))
        first = {
            m.split("turn=")[1].split()[0]
            for m in (r.getMessage() for r in caplog.records)
            if "turn=" in m
        }
        caplog.clear()
        drain(app.respond("two", [], app.DEFAULT_MODEL))
        second = {
            m.split("turn=")[1].split()[0]
            for m in (r.getMessage() for r in caplog.records)
            if "turn=" in m
        }

    assert len(first) == 1 and len(second) == 1, "one turn must use one id"
    assert first != second, "consecutive turns must not share an id"


def test_rejected_login_is_logged_without_the_password(caplog):
    with caplog.at_level("INFO", logger="gemma-endpoint"):
        assert app.verify_login("intruder", "hunter2") is False

    record = caplog.records[-1]
    assert record.levelname == "WARNING"
    assert "REJECTED" in record.getMessage()
    assert "hunter2" not in record.getMessage()
    assert "s3cret" not in record.getMessage()


def test_logs_never_contain_the_hf_token(monkeypatch, caplog):
    """The token is only ever an Authorization header, never a log field."""
    response = FakeStreamResponse(sse(chunk("hi"), "[DONE]"))
    monkeypatch.setattr(app.requests, "post", lambda url, **kwargs: response)

    with caplog.at_level("DEBUG", logger="gemma-endpoint"):
        for model in app.AVAILABLE_MODELS:
            list(
                app.stream_completion(
                    [{"role": "user", "content": "hi"}], model, turn=1
                )
            )

    assert caplog.records, "expected the request to be logged"
    tokens = {endpoint.token for endpoint in app.MODELS.values()}
    for record in caplog.records:
        assert not any(token in record.getMessage() for token in tokens)
        assert "Authorization" not in record.getMessage()


def test_log_level_applies_to_this_app_only():
    """LOG_LEVEL must not switch on httpx/urllib3 chatter along with ours."""
    import logging

    assert app.logger.level == logging.INFO
    assert logging.getLogger().level == logging.WARNING
    assert app.logger.isEnabledFor(logging.INFO)
    assert not logging.getLogger("httpx").isEnabledFor(logging.INFO)


@pytest.fixture
def no_sleep(monkeypatch):
    """Run retry loops at full speed, with a clock that still advances."""
    clock = {"now": 0.0}

    monkeypatch.setattr(app.time, "monotonic", lambda: clock["now"])
    monkeypatch.setattr(
        app.time, "sleep", lambda seconds: clock.update(now=clock["now"] + seconds)
    )

    return clock


@pytest.mark.parametrize("status", [502, 503, 504])
def test_stream_completion_flags_a_cold_endpoint(monkeypatch, status):
    monkeypatch.setattr(
        app.requests, "post", lambda url, **kw: FakeStreamResponse([], status)
    )

    with pytest.raises(app.EndpointWaking):
        list(app.stream_completion([], "m"))


def test_stream_completion_still_raises_for_real_errors(monkeypatch):
    response = FakeStreamResponse([], 404)

    def raise_for_status():
        raise requests.exceptions.HTTPError("404", response=None)

    response.raise_for_status = raise_for_status
    monkeypatch.setattr(app.requests, "post", lambda url, **kw: response)

    with pytest.raises(requests.exceptions.HTTPError):
        list(app.stream_completion([], "m"))


def test_stream_turn_waits_out_a_cold_endpoint(monkeypatch, no_sleep):
    attempts = {"n": 0}

    def stream(messages, model_name, turn="-"):
        attempts["n"] += 1

        if attempts["n"] < 3:
            raise app.EndpointWaking(503)
            yield  # pragma: no cover

        yield "awake"

    monkeypatch.setattr(app, "stream_completion", stream)

    events = list(app.stream_turn([], "m", turn=1))

    assert attempts["n"] == 3, "should retry until the endpoint answers"
    assert ("token", "awake") in events
    waking = [value for kind, value in events if kind == "waking"]
    assert waking, "no waking updates were emitted"
    assert waking == sorted(waking), "the waited counter must not go backwards"


def test_stream_turn_gives_up_after_the_wake_timeout(monkeypatch, no_sleep):
    import gradio as gr

    def always_cold(messages, model_name, turn="-"):
        raise app.EndpointWaking(503)
        yield  # pragma: no cover

    monkeypatch.setattr(app, "stream_completion", always_cold)

    with pytest.raises(gr.Error, match="did not come up"):
        for _ in app.stream_turn([], "m", turn=1):
            pass

    assert no_sleep["now"] >= app.WAKE_TIMEOUT_SECONDS


def test_respond_shows_a_waking_message_then_the_reply(monkeypatch, no_sleep):
    attempts = {"n": 0}

    def stream(messages, model_name, turn="-"):
        attempts["n"] += 1

        if attempts["n"] == 1:
            raise app.EndpointWaking(503)
            yield  # pragma: no cover

        yield "hello"

    monkeypatch.setattr(app, "stream_completion", stream)

    frames, final = drain(app.respond("ping", [], app.DEFAULT_MODEL))
    titles = [
        (h[-1].get("metadata") or {}).get("title") for h in histories(frames) if h
    ]

    assert any(t and "Waking up the endpoint" in t for t in titles)
    assert all(locked(f) for f in frames[:-1]), "controls stay locked while waking"
    assert final[0][-1] == {"role": "assistant", "content": "hello"}


def test_respond_reports_a_cold_endpoint_that_never_wakes(monkeypatch, no_sleep):
    import gradio as gr

    def always_cold(messages, model_name, turn="-"):
        raise app.EndpointWaking(503)
        yield  # pragma: no cover

    monkeypatch.setattr(app, "stream_completion", always_cold)

    frames = drain_failing(app.respond("ping", [], app.DEFAULT_MODEL), gr.Error)

    assert frames[-1][0] == [], "the failed turn is rolled back"
    assert textbox(frames[-1]) == "ping"
    assert locked(frames[-1]) is False


class Utf8SseServer:
    """A real HTTP server streaming raw UTF-8 SSE, as vLLM/llama.cpp do.

    `FakeStreamResponse` hands back `str` lines that are already decoded, so it
    cannot catch a decoding bug. Only a real socket exercises the path where
    `requests` has to pick a charset.
    """

    def __init__(self, content_type):
        import http.server
        import threading

        body = "".join(
            f"data: {json.dumps({'choices': [{'delta': {'content': token}}]}, ensure_ascii=False)}\n\n"
            for token in ("こんにちは", "、", "日本語", "です", "。")
        )
        raw = (body + "data: [DONE]\n\n").encode("utf-8")

        class Handler(http.server.BaseHTTPRequestHandler):
            def do_POST(self):
                self.rfile.read(int(self.headers.get("Content-Length") or 0))
                self.send_response(200)
                self.send_header("Content-Type", content_type)
                self.send_header("Transfer-Encoding", "chunked")
                self.end_headers()

                # Split on a 7-byte boundary so multi-byte characters land
                # across two chunks, as they do on a real connection.
                for start in range(0, len(raw), 7):
                    piece = raw[start : start + 7]
                    self.wfile.write(b"%x\r\n%s\r\n" % (len(piece), piece))
                    self.wfile.flush()

                self.wfile.write(b"0\r\n\r\n")

            def log_message(self, *args):
                pass

        self._server = http.server.HTTPServer(("127.0.0.1", 0), Handler)
        threading.Thread(target=self._server.serve_forever, daemon=True).start()

    @property
    def url(self):
        return f"http://127.0.0.1:{self._server.server_port}"

    def close(self):
        self._server.shutdown()


@pytest.mark.parametrize(
    "content_type",
    ["text/event-stream", "text/event-stream; charset=utf-8"],
)
def test_stream_completion_decodes_utf8_replies(monkeypatch, content_type):
    """Regression: a charset-less SSE response decoded Japanese as mojibake.

    `requests` falls back to ISO-8859-1 for a text/* response with no charset,
    so every non-ASCII reply came back garbled while English looked fine.
    """
    server = Utf8SseServer(content_type)
    monkeypatch.setattr(app, "ENDPOINT_URL", server.url)

    try:
        reply = "".join(app.stream_completion([], "m"))
    finally:
        server.close()

    assert reply == "こんにちは、日本語です。"