"""A model picked mid-turn must still get its selection-guard confirm step.

``config.set model`` on a *running* session cannot swap the agent in place --
the worker thread is reading ``agent.model`` / ``agent.client`` on every
iteration -- so it stashes the pick in ``session["pending_model_switch"]`` and
``_apply_pending_model_switch`` applies it at the next turn start.

That deferral used to skip the selection guards entirely: the stash branch
answered ``confirm_required: False`` without ever calling them. A client that
implements the confirm round-trip was therefore told no consent was needed, so
it never prompted. One turn later ``_apply_pending_model_switch`` ran the
guards with the stashed (unconfirmed) flag, saw the warning, and deliberately
dropped the switch -- correct on its own terms, but by then no round-trip was
possible. The user's pick silently reverted and the confirm was never offered
on this path at all.
"""

import logging
import threading
import types

import pytest

from tui_gateway import server

# A vendor-documented data-training tier. The data-policy guard keys on the
# model id alone (no base_url / api_key / model_info), which is exactly what
# the stash branch can see before resolution.
GUARDED_MODEL = "muse-spark-1.2-contributor"
UNGUARDED_MODEL = "anthropic/claude-sonnet-4.6"

def _session(**extra):
    return {
        "agent": types.SimpleNamespace(),
        "session_key": "session-key",
        "history": [],
        "history_lock": threading.Lock(),
        "history_version": 0,
        "running": False,
        "attached_images": [],
        "image_counter": 0,
        "cols": 80,
        "slash_worker": None,
        "show_reasoning": False,
        "tool_progress_mode": "all",
        **extra,
    }

def _config_set_model(value, **extra_params):
    params = {"session_id": "sid", "key": "model", "value": value}
    params.update(extra_params)

    return server.handle_request({"id": "1", "method": "config.set", "params": params})

@pytest.fixture
def running_session(monkeypatch):
    """A busy session whose live swap path is fatal if it is ever reached."""

    def _must_not_run(*_args, **_kwargs):
        raise AssertionError(
            "_apply_model_switch ran on the busy path -- it would race the "
            "worker thread reading agent.model / agent.client"
        )

    monkeypatch.setattr(server, "_apply_model_switch", _must_not_run)
    server._sessions["sid"] = _session(running=True)
    try:
        yield server._sessions["sid"]
    finally:
        server._sessions.pop("sid", None)

class TestGuardedPickAsksBeforeStashing:
    def test_reports_confirm_required_instead_of_deferring(self, running_session):
        resp = _config_set_model(GUARDED_MODEL)

        assert not resp.get("error")
        result = resp["result"]
        assert result["confirm_required"] is True, (
            "the deferred path answered confirm_required=False without running "
            "the guards, so a correct client never prompts and the pick is "
            "dropped a turn later with no way to consent"
        )
        assert result["confirm_message"].strip()
        assert result["deferred"] is False

    def test_leaves_the_session_untouched(self, running_session):
        _config_set_model(GUARDED_MODEL)

        assert "pending_model_switch" not in running_session, (
            "an unconfirmed guarded pick must not be queued -- the next turn "
            "start would drop it anyway, after the pill already moved"
        )

    def test_reconfirming_queues_the_pick(self, running_session):
        resp = _config_set_model(GUARDED_MODEL, confirm_expensive_model=True)

        result = resp["result"]
        assert result["deferred"] is True
        assert result["confirm_required"] is False

        pending = running_session["pending_model_switch"]
        assert pending["raw"] == GUARDED_MODEL
        assert pending["confirm_expensive_model"] is True, (
            "the ack must survive into the stash or _apply_pending_model_switch "
            "re-runs the guard at turn start and drops the confirmed pick"
        )

class TestUnguardedPickStillDefers:
    """The queue-don't-race behaviour is the whole point of this branch."""

    def test_defers_without_a_confirm_step(self, running_session):
        result = _config_set_model(UNGUARDED_MODEL)["result"]

        assert result["deferred"] is True
        assert result["confirm_required"] is False
        assert result["confirm_message"] == ""
        assert result["value"] == UNGUARDED_MODEL

    def test_stashes_the_pick_for_the_next_turn(self, running_session):
        _config_set_model(UNGUARDED_MODEL)

        pending = running_session["pending_model_switch"]
        assert pending["raw"] == UNGUARDED_MODEL
        assert pending["confirm_expensive_model"] is False

    def test_explicit_provider_is_still_recorded_for_display(self, running_session):
        _config_set_model(f"{UNGUARDED_MODEL} --provider anthropic")

        pending = running_session["pending_model_switch"]
        assert pending["display_provider"] == "anthropic"

class TestGuardFailureIsNotFatal:
    def test_a_raising_guard_falls_back_to_deferring(self, running_session, monkeypatch):
        """A broken guard must never cost the user their model pick.

        The apply-time check in ``_apply_pending_model_switch`` is still there,
        so failing open here degrades to the old behaviour rather than to a
        silently unguarded switch.
        """

        def _boom(*_args, **_kwargs):
            raise RuntimeError("guard table is broken")

        monkeypatch.setattr(
            "hermes_cli.model_selection_guards.combined_selection_warning", _boom
        )

        result = _config_set_model(GUARDED_MODEL)["result"]

        assert result["deferred"] is True
        assert running_session["pending_model_switch"]["raw"] == GUARDED_MODEL

@pytest.fixture
def threshold():
    from hermes_cli.model_selection_guards import _context_cache_threshold

    return _context_cache_threshold()

def _live_agent(session, tokens, model="deepseek/deepseek-v4.1-flash"):
    session["agent"] = types.SimpleNamespace(
        model=model, context_compressor=types.SimpleNamespace(last_prompt_tokens=tokens))

class TestLargeContextPickAsksBeforeStashing:
    def test_session_at_the_threshold_is_asked_at_pick_time(self, running_session, threshold):
        _live_agent(running_session, threshold)

        result = _config_set_model(UNGUARDED_MODEL)["result"]

        assert result["confirm_required"] is True, (
            "the context-cache guard only saw the session size at turn start, where it can no longer "
            "ask: the pick became an error toast and was dropped"
        )
        assert "LARGE CONTEXT MODEL SWITCH" in result["confirm_message"]
        assert result["deferred"] is False
        assert "pending_model_switch" not in running_session

    def test_confirming_queues_the_pick(self, running_session, threshold):
        _live_agent(running_session, threshold)

        result = _config_set_model(UNGUARDED_MODEL, confirm_expensive_model=True)["result"]

        assert result["deferred"] is True
        assert running_session["pending_model_switch"]["confirm_expensive_model"] is True

    def test_session_under_the_threshold_defers_without_asking(self, running_session, threshold):
        _live_agent(running_session, threshold - 1)

        result = _config_set_model(UNGUARDED_MODEL)["result"]

        assert result["deferred"] is True
        assert result["confirm_required"] is False

    def test_reselecting_the_live_model_is_not_a_switch(self, running_session, threshold):
        _live_agent(running_session, threshold, model=UNGUARDED_MODEL)

        result = _config_set_model(UNGUARDED_MODEL)["result"]

        assert result["confirm_required"] is False

    def test_no_live_agent_defers_without_asking(self, running_session):
        running_session["agent"] = None

        result = _config_set_model(UNGUARDED_MODEL)["result"]

        assert result["deferred"] is True
        assert result["confirm_required"] is False
        assert running_session["pending_model_switch"]["raw"] == UNGUARDED_MODEL

def test_bare_pick_is_guarded_against_the_live_provider(running_session, monkeypatch):
    """A pick without --provider resolves against the live provider at turn start, so a
    provider-keyed price check must see that provider at pick time or it drops the pick later."""
    from hermes_cli import model_cost_guard

    monkeypatch.setattr(model_cost_guard, "expensive_model_warning", lambda model, provider=None, **_kw: (
        types.SimpleNamespace(message="pricey") if provider == "openai-api" else None))
    running_session["agent"] = types.SimpleNamespace(model="gpt-4.1-nano", provider="openai-api")

    result = _config_set_model("o1-pro")["result"]

    assert result["confirm_required"] is True and result["deferred"] is False
    assert "pending_model_switch" not in running_session

class TestDroppedQueuedPickIsLogged:
    def test_turn_start_drop_leaves_a_server_log_line(self, monkeypatch, caplog):
        session = _session(pending_model_switch={
            "raw": UNGUARDED_MODEL, "confirm_expensive_model": False,
            "display_model": UNGUARDED_MODEL, "display_provider": ""})
        monkeypatch.setattr(
            server, "_apply_model_switch",
            lambda *_a, **_kw: {"confirm_required": True, "confirm_message": "guarded"})
        emitted = []
        monkeypatch.setattr(server, "_emit", lambda *a, **_kw: emitted.append(a))

        with caplog.at_level(logging.WARNING):
            server._apply_pending_model_switch("sid", session)

        assert "dropped" in caplog.text and UNGUARDED_MODEL in caplog.text, (
            "a dropped pick used to exist only as a client toast, invisible in server logs"
        )
        assert "pending_model_switch" not in session


class TestDroppedQueuedPickIsNotATurnError:
    """A pick dropped at turn start must not fail the turn: Desktop paints ``error`` as the red retry card
    under the user's message, while the turn itself runs on the old model."""

    def _run(self, monkeypatch, apply):
        session = _session(pending_model_switch={
            "raw": UNGUARDED_MODEL, "confirm_expensive_model": False,
            "display_model": UNGUARDED_MODEL, "display_provider": ""})
        session["agent"] = types.SimpleNamespace(model="deepseek/deepseek-v4.1-flash")
        monkeypatch.setattr(server, "_apply_model_switch", apply)
        emitted = []
        monkeypatch.setattr(server, "_emit", lambda *a, **_kw: emitted.append(a))
        server._apply_pending_model_switch("sid", session)
        return [a[0] for a in emitted], emitted

    def test_guarded_drop_is_a_warning_and_resyncs_the_pill(self, monkeypatch):
        kinds, emitted = self._run(
            monkeypatch, lambda *_a, **_kw: {"confirm_required": True, "confirm_message": "guarded"})

        assert "error" not in kinds
        notice = next(a[2] for a in emitted if a[0] == "notification.show")
        assert notice["level"] == "warn" and UNGUARDED_MODEL in notice["text"]
        assert "\n" not in notice["text"], "the guard banner belongs in the re-pick confirm, not the toast"
        assert "session.info" in kinds

    def test_failed_switch_is_a_warning(self, monkeypatch):
        def _boom(*_a, **_kw):
            raise ValueError("no credentials")

        kinds, _ = self._run(monkeypatch, _boom)

        assert "error" not in kinds
        assert "notification.show" in kinds