vrobbler/tests/agents_tests/test_agent_sessions.py

import json
import subprocess
from unittest.mock import MagicMock, patch

import httpx
import pytest
from agents.models import AgentSession, AgentSessionLogData
from agents.providers import (
    _parse_opencode_events,
    agent_prompt,
    friendly_error_message,
    gemini_agent_prompt,
    is_transient_error,
    list_gemini_agent_models,
    list_openrouter_free_models,
    opencode_agent_prompt,
    openrouter_agent_prompt,
    retry_after_from,
)
from django.conf import settings
from django.contrib.auth import get_user_model
from django.urls import reverse
from django.utils import timezone
from scrobbles.models import Scrobble
from scrobbles.notifications import (
    AgentSessionFailedNtfyNotification,
    AgentSessionResponseNtfyNotification,
)
from scrobbles.scrobblers import (
    manual_scrobble_agent_follow_up,
    manual_scrobble_agent_session,
)
from scrobbles.tasks import (
    AGENT_SESSION_AUTO_COMPLETE_SECONDS,
    scrobble_agent_session_complete,
    scrobble_agent_session_prompt,
)

User = get_user_model()


@pytest.fixture
def user(db):
    return User.objects.create_user(username="agentuser", password="pw")


def _mk_scrobble(user, *, in_progress=True, turns=None, title="First prompt"):
    agent_session, _ = AgentSession.find_or_create(
        provider=settings.AGENT_PROVIDER, model=settings.LLM_MODEL
    )
    scrobble = Scrobble.create_or_update(
        agent_session,
        user.id,
        {
            "user_id": user.id,
            "timestamp": timezone.now(),
            "playback_position_seconds": 0,
            "source": "Vrobbler",
        },
        skip_in_progress_check=True,
    )
    log = scrobble.log if isinstance(scrobble.log, dict) else {}
    log["provider"] = settings.AGENT_PROVIDER
    log["model"] = settings.LLM_MODEL
    log["title"] = title
    log["turns"] = turns or []
    scrobble.log = log
    scrobble.in_progress = in_progress
    scrobble.save(update_fields=["log", "in_progress"])
    return scrobble


# --- providers ---


def test_parse_opencode_events_joins_text_parts():
    raw = "\n".join(
        [
            json.dumps({"type": "step_start"}),
            json.dumps({"type": "text", "part": {"type": "text", "text": "hello"}}),
            json.dumps({"type": "text", "part": {"type": "text", "text": " world"}}),
            json.dumps({"type": "step_finish"}),
            "not-json",
        ]
    )
    assert _parse_opencode_events(raw) == "hello\n world"


def test_opencode_agent_prompt(settings):
    settings.OPENCODE_CMD = "opencode"
    events = json.dumps(
        {"type": "text", "part": {"type": "text", "text": "the answer"}}
    )
    with patch(
        "agents.providers.subprocess.run",
        return_value=MagicMock(returncode=0, stdout=events, stderr=""),
    ) as mock_run:
        result = opencode_agent_prompt("what is 2+2?")

    mock_run.assert_called_once()
    cmd = mock_run.call_args.args[0]
    assert cmd[:2] == ["opencode", "run"]
    assert cmd[2:5] == ["--format", "json", "--dir"]
    assert cmd[5].startswith("/tmp/")
    assert cmd[6] == "what is 2+2?"
    assert result == {"text": "the answer", "provider": "opencode", "model": "opencode"}


def test_opencode_agent_prompt_nonzero_exit():
    with patch(
        "agents.providers.subprocess.run",
        return_value=MagicMock(returncode=1, stdout="", stderr="boom"),
    ):
        with pytest.raises(ValueError, match="opencode exited with 1"):
            opencode_agent_prompt("hi")


def test_opencode_agent_prompt_timeout():
    with patch(
        "agents.providers.subprocess.run",
        side_effect=subprocess.TimeoutExpired("cmd", 900),
    ):
        with pytest.raises(ValueError, match="timed out"):
            opencode_agent_prompt("hi")


def test_opencode_agent_prompt_no_text_output():
    with patch(
        "agents.providers.subprocess.run",
        return_value=MagicMock(
            returncode=0, stdout=json.dumps({"type": "step_start"}), stderr=""
        ),
    ):
        with pytest.raises(ValueError, match="no text output"):
            opencode_agent_prompt("hi")


def test_gemini_agent_prompt(settings):
    settings.GOOGLE_AI_API_KEY = "test-key"
    settings.LLM_MODEL = "gemini-2.5-flash"
    fake = MagicMock()
    fake.raise_for_status = MagicMock()
    fake.json.return_value = {
        "candidates": [{"content": {"parts": [{"text": "  the answer  "}]}}]
    }
    with patch("agents.providers.httpx.post", return_value=fake) as mock_post:
        result = gemini_agent_prompt("what is 2+2?")

    mock_post.assert_called_once()
    kwargs = mock_post.call_args.kwargs
    assert kwargs["params"] == {"key": "test-key"}
    assert kwargs["json"] == {"contents": [{"parts": [{"text": "what is 2+2?"}]}]}
    assert result == {
        "text": "the answer",
        "provider": "gemini",
        "model": "gemini-2.5-flash",
    }


def test_gemini_agent_prompt_missing_key(settings):
    settings.GOOGLE_AI_API_KEY = ""
    with pytest.raises(ValueError, match="VROBBLER_GOOGLE_AI_API_KEY is not set"):
        gemini_agent_prompt("hi")


def test_gemini_agent_prompt_bad_response(settings):
    settings.GOOGLE_AI_API_KEY = "test-key"
    fake = MagicMock()
    fake.raise_for_status = MagicMock()
    fake.json.return_value = {"unexpected": True}
    with patch("agents.providers.httpx.post", return_value=fake):
        with pytest.raises(ValueError, match="Unexpected Gemini response"):
            gemini_agent_prompt("hi")


def test_gemini_agent_prompt_with_history(settings):
    settings.GOOGLE_AI_API_KEY = "test-key"
    fake = MagicMock()
    fake.raise_for_status = MagicMock()
    fake.json.return_value = {"candidates": [{"content": {"parts": [{"text": "12"}]}}]}
    history = [
        {"prompt": "what is 2+2?", "response": "4"},
        {"prompt": "no response yet", "response": None},
    ]
    with patch("agents.providers.httpx.post", return_value=fake) as mock_post:
        gemini_agent_prompt("times 3?", history=history)

    contents = mock_post.call_args.kwargs["json"]["contents"]
    assert contents == [
        {"parts": [{"text": "what is 2+2?"}]},
        {"parts": [{"text": "4"}]},
        {"parts": [{"text": "times 3?"}]},
    ]


def test_opencode_agent_prompt_with_history(settings):
    settings.OPENCODE_CMD = "opencode"
    with patch(
        "agents.providers.subprocess.run",
        return_value=MagicMock(
            returncode=0,
            stdout=json.dumps({"type": "text", "part": {"type": "text", "text": "x"}}),
            stderr="",
        ),
    ) as mock_run:
        opencode_agent_prompt(
            "times 3?", history=[{"prompt": "what is 2+2?", "response": "4"}]
        )

    prompt = mock_run.call_args.args[0][6]
    assert "Previous conversation:" in prompt
    assert "User: what is 2+2?" in prompt
    assert "Assistant: 4" in prompt
    assert "times 3?" in prompt


def test_agent_prompt_unknown_provider(settings):
    settings.AGENT_PROVIDER = "nope"
    with pytest.raises(ValueError, match="Unknown agent provider"):
        agent_prompt("hi")


def test_agent_prompt_defaults_to_configured_provider(settings):
    settings.AGENT_PROVIDER = "gemini"
    settings.GOOGLE_AI_API_KEY = "test-key"
    with patch("agents.providers.gemini_agent_prompt", return_value={"text": "x"}) as m:
        agent_prompt("hi")
    m.assert_called_once_with("hi", history=None, model=None)


def test_agent_prompt_openrouter_dispatches(settings):
    settings.AGENT_PROVIDER = "openrouter"
    settings.OPENROUTER_API_KEY = "sk-test"
    with patch(
        "agents.providers.openrouter_agent_prompt", return_value={"text": "x"}
    ) as m:
        agent_prompt("hi", model="free/model-a")
    m.assert_called_once_with("hi", history=None, model="free/model-a")


def test_openrouter_agent_prompt(settings):
    settings.OPENROUTER_API_KEY = "sk-test"
    settings.LLM_MODEL = "some/model"
    fake = MagicMock()
    fake.raise_for_status = MagicMock()
    fake.json.return_value = {"choices": [{"message": {"content": "  the answer  "}}]}
    with patch("agents.providers.httpx.post", return_value=fake) as mock_post:
        result = openrouter_agent_prompt("what is 2+2?")

    mock_post.assert_called_once()
    kwargs = mock_post.call_args.kwargs
    assert kwargs["headers"] == {"Authorization": "Bearer sk-test"}
    assert kwargs["json"] == {
        "model": "some/model",
        "messages": [{"role": "user", "content": "what is 2+2?"}],
    }
    assert result == {
        "text": "the answer",
        "provider": "openrouter",
        "model": "some/model",
    }


def test_openrouter_agent_prompt_missing_key(settings):
    settings.OPENROUTER_API_KEY = ""
    with pytest.raises(ValueError, match="VROBBLER_OPENROUTER_API_KEY is not set"):
        openrouter_agent_prompt("hi")


def test_openrouter_agent_prompt_bad_response(settings):
    settings.OPENROUTER_API_KEY = "sk-test"
    fake = MagicMock()
    fake.raise_for_status = MagicMock()
    fake.json.return_value = {"unexpected": True}
    with patch("agents.providers.httpx.post", return_value=fake):
        with pytest.raises(ValueError, match="Unexpected OpenRouter response"):
            openrouter_agent_prompt("hi")


def test_openrouter_agent_prompt_with_history(settings):
    settings.OPENROUTER_API_KEY = "sk-test"
    fake = MagicMock()
    fake.raise_for_status = MagicMock()
    fake.json.return_value = {"choices": [{"message": {"content": "12"}}]}
    history = [
        {"prompt": "what is 2+2?", "response": "4"},
        {"prompt": "no response yet", "response": None},
    ]
    with patch("agents.providers.httpx.post", return_value=fake) as mock_post:
        openrouter_agent_prompt("times 3?", history=history, model="some/model")

    messages = mock_post.call_args.kwargs["json"]["messages"]
    assert messages == [
        {"role": "user", "content": "what is 2+2?"},
        {"role": "assistant", "content": "4"},
        {"role": "user", "content": "times 3?"},
    ]


def test_list_openrouter_free_models_filters_free():
    fake = MagicMock()
    fake.raise_for_status = MagicMock()
    fake.json.return_value = {
        "data": [
            {
                "id": "paid/model",
                "name": "Paid",
                "pricing": {"prompt": "0.001", "completion": "0.002"},
            },
            {
                "id": "free/model-a",
                "name": "Bee",
                "pricing": {"prompt": "0", "completion": "0"},
            },
            {
                "id": "free/model-b",
                "name": "Alpha",
                "pricing": {"prompt": "0", "completion": "0"},
            },
        ]
    }
    with patch("agents.providers.httpx.get", return_value=fake) as mock_get:
        models = list_openrouter_free_models()

    mock_get.assert_called_once()
    assert models == [
        {"provider": "openrouter", "id": "free/model-b", "name": "Alpha"},
        {"provider": "openrouter", "id": "free/model-a", "name": "Bee"},
    ]


def test_list_openrouter_free_models_skips_missing_pricing():
    fake = MagicMock()
    fake.raise_for_status = MagicMock()
    fake.json.return_value = {
        "data": [
            {"id": "no-pricing/model", "name": "No Pricing"},
        ]
    }
    with patch("agents.providers.httpx.get", return_value=fake):
        assert list_openrouter_free_models() == []


def test_list_gemini_agent_models(settings):
    settings.GOOGLE_AI_API_KEY = "test-key"
    settings.GEMINI_AGENT_MODELS = "gemini-2.5-flash, gemini-3.6-flash"
    assert list_gemini_agent_models() == [
        {
            "provider": "gemini",
            "id": "gemini-2.5-flash",
            "name": "gemini-2.5-flash",
        },
        {
            "provider": "gemini",
            "id": "gemini-3.6-flash",
            "name": "gemini-3.6-flash",
        },
    ]


def test_list_gemini_agent_models_no_key(settings):
    settings.GOOGLE_AI_API_KEY = ""
    assert list_gemini_agent_models() == []


# --- retry helpers ---


def _http_error(status_code, headers=None):
    request = httpx.Request("POST", "https://openrouter.ai/api/v1/chat/completions")
    response = httpx.Response(status_code, request=request, headers=headers or {})
    return httpx.HTTPStatusError("provider error", request=request, response=response)


def test_is_transient_error_429():
    assert is_transient_error(_http_error(429)) is True


def test_is_transient_error_5xx():
    assert is_transient_error(_http_error(500)) is True
    assert is_transient_error(_http_error(502)) is True
    assert is_transient_error(_http_error(503)) is True
    assert is_transient_error(_http_error(504)) is True


def test_is_transient_error_4xx_not_retryable():
    assert is_transient_error(_http_error(400)) is False
    assert is_transient_error(_http_error(401)) is False
    assert is_transient_error(_http_error(403)) is False


def test_is_transient_error_transport():
    assert is_transient_error(httpx.ConnectError("boom")) is True
    assert is_transient_error(httpx.ReadTimeout("boom")) is True


def test_is_transient_error_plain_exception():
    assert is_transient_error(ValueError("boom")) is False


def test_retry_after_from_header():
    assert retry_after_from(_http_error(429, {"Retry-After": "45"})) == 45


def test_retry_after_from_missing():
    assert retry_after_from(_http_error(429)) is None
    assert retry_after_from(ValueError("boom")) is None


def test_retry_after_from_invalid_header():
    assert retry_after_from(_http_error(429, {"Retry-After": "soon"})) is None


def test_friendly_error_message_http():
    msg = friendly_error_message(_http_error(429))
    assert msg == "Provider returned HTTP 429 (Too Many Requests)"


def test_friendly_error_message_plain():
    assert friendly_error_message(ValueError("boom")) == "boom"


# --- scrobbler ---


@patch("scrobbles.tasks.scrobble_agent_session_prompt.delay")
def test_manual_scrobble_agent_session_creates_new_session(mock_delay, user):
    scrobble = manual_scrobble_agent_session("What is the capital of France?", user.id)

    assert Scrobble.objects.filter(id=scrobble.id).exists()
    assert scrobble.media_type == Scrobble.MediaType.AGENT_SESSION
    assert scrobble.in_progress is True
    assert scrobble.agent_session.provider == settings.AGENT_PROVIDER
    assert scrobble.agent_session.model == settings.LLM_MODEL
    assert (
        scrobble.agent_session.title
        == f"{settings.AGENT_PROVIDER} {settings.LLM_MODEL}"
    )
    assert scrobble.log["title"] == "What is the capital of France?"
    turns = scrobble.log["turns"]
    assert len(turns) == 1
    assert turns[0]["prompt"] == "What is the capital of France?"
    assert turns[0]["response"] is None
    mock_delay.assert_called_once_with(scrobble.id, turns[0]["prompt_id"])


@patch("scrobbles.tasks.scrobble_agent_session_prompt.delay")
def test_manual_scrobble_agent_session_reuses_agent_session(mock_delay, user):
    agent_session, created = AgentSession.find_or_create(
        provider=settings.AGENT_PROVIDER, model=settings.LLM_MODEL
    )
    assert created is True

    manual_scrobble_agent_session("First question?", user.id)

    agent_session_again, created = AgentSession.find_or_create(
        provider=settings.AGENT_PROVIDER, model=settings.LLM_MODEL
    )
    assert created is False
    assert agent_session_again.id == agent_session.id
    assert AgentSession.objects.count() == 1


@patch("scrobbles.tasks.scrobble_agent_session_prompt.delay")
def test_manual_scrobble_agent_session_appends_to_in_progress(mock_delay, user):
    _mk_scrobble(
        user,
        in_progress=True,
        turns=[
            {"prompt_id": "first-1", "prompt": "First prompt", "response": "answer"}
        ],
    )

    scrobble = manual_scrobble_agent_session("And the second question?", user.id)

    turns = scrobble.log["turns"]
    assert len(turns) == 2
    assert turns[0]["prompt"] == "First prompt"
    assert turns[1]["prompt"] == "And the second question?"
    assert turns[1]["response"] is None
    assert scrobble.log["title"] == "First prompt"


@patch("scrobbles.tasks.scrobble_agent_session_prompt.delay")
def test_manual_scrobble_agent_session_new_session_after_completed(mock_delay, user):
    _mk_scrobble(user, in_progress=False)

    scrobble = manual_scrobble_agent_session("A new session?", user.id)

    assert scrobble.media_type == Scrobble.MediaType.AGENT_SESSION
    assert len(scrobble.log["turns"]) == 1
    assert Scrobble.objects.filter(user_id=user.id).count() == 2


@patch("scrobbles.tasks.scrobble_agent_session_prompt.delay")
def test_manual_scrobble_agent_session_opencode_model_empty(mock_delay, user, settings):
    settings.AGENT_PROVIDER = "opencode"

    scrobble = manual_scrobble_agent_session("hi", user.id)

    assert scrobble.agent_session.provider == "opencode"
    assert scrobble.agent_session.model == "opencode"
    assert scrobble.agent_session.title == "opencode opencode"


# --- celery task ---


@patch("agents.providers.agent_prompt")
@patch("scrobbles.tasks.scrobble_agent_session_complete.apply_async")
def test_scrobble_agent_session_prompt_fills_turn(
    mock_complete, mock_agent_prompt, user
):
    scrobble = _mk_scrobble(
        user,
        turns=[
            {
                "prompt_id": "abc-123",
                "prompt": "hello",
                "response": None,
            }
        ],
    )
    mock_agent_prompt.return_value = {
        "text": "hi back",
        "provider": settings.AGENT_PROVIDER,
        "model": settings.LLM_MODEL,
    }

    scrobble_agent_session_prompt(scrobble.id, "abc-123")

    scrobble.refresh_from_db()
    mock_agent_prompt.assert_called_once_with(
        "hello", provider=settings.AGENT_PROVIDER, history=[], model=settings.LLM_MODEL
    )
    assert scrobble.log["turns"][0]["response"] == "hi back"
    assert scrobble.log["provider"] == settings.AGENT_PROVIDER
    assert scrobble.in_progress is True
    assert scrobble.played_to_completion is False
    mock_complete.assert_called_once_with(
        args=[scrobble.id], countdown=AGENT_SESSION_AUTO_COMPLETE_SECONDS
    )


@patch("agents.providers.agent_prompt")
@patch("scrobbles.tasks.scrobble_agent_session_complete.apply_async")
def test_scrobble_agent_session_prompt_passes_history(
    mock_complete, mock_agent_prompt, user
):
    scrobble = _mk_scrobble(
        user,
        turns=[
            {
                "prompt_id": "first-1",
                "prompt": "what is 2+2?",
                "response": "4",
            },
            {
                "prompt_id": "err-1",
                "prompt": "boom?",
                "response": "Error: boom",
                "error": True,
            },
            {
                "prompt_id": "cur-1",
                "prompt": "and times 3?",
                "response": None,
            },
        ],
    )

    mock_agent_prompt.return_value = {
        "text": "12",
        "provider": settings.AGENT_PROVIDER,
        "model": settings.LLM_MODEL,
    }
    scrobble_agent_session_prompt(scrobble.id, "cur-1")

    mock_agent_prompt.assert_called_once_with(
        "and times 3?",
        provider=settings.AGENT_PROVIDER,
        history=[{"prompt": "what is 2+2?", "response": "4"}],
        model=settings.LLM_MODEL,
    )


@patch("agents.providers.agent_prompt")
@patch("scrobbles.tasks.scrobble_agent_session_complete.apply_async")
def test_scrobble_agent_session_prompt_stores_error(
    mock_complete, mock_agent_prompt, user
):
    scrobble = _mk_scrobble(
        user,
        turns=[
            {
                "prompt_id": "abc-123",
                "prompt": "hello",
                "response": None,
            }
        ],
    )
    mock_agent_prompt.side_effect = ValueError("boom")

    scrobble_agent_session_prompt(scrobble.id, "abc-123")

    scrobble.refresh_from_db()
    assert scrobble.log["turns"][0]["response"] == "Error: boom"
    assert scrobble.log["turns"][0]["error"] is True
    assert scrobble.in_progress is True
    mock_complete.assert_called_once_with(
        args=[scrobble.id], countdown=AGENT_SESSION_AUTO_COMPLETE_SECONDS
    )


@patch("agents.providers.agent_prompt")
@patch("scrobbles.notifications.AgentSessionResponseNtfyNotification")
@patch("scrobbles.tasks.scrobble_agent_session_complete.apply_async")
def test_scrobble_agent_session_prompt_sends_ntfy_on_response(
    mock_complete, mock_notification, mock_agent_prompt, user
):
    scrobble = _mk_scrobble(
        user,
        turns=[
            {
                "prompt_id": "abc-123",
                "prompt": "hello",
                "response": None,
            }
        ],
    )
    mock_agent_prompt.return_value = {
        "text": "hi back",
        "provider": settings.AGENT_PROVIDER,
        "model": settings.LLM_MODEL,
    }

    scrobble_agent_session_prompt(scrobble.id, "abc-123")

    mock_notification.assert_called_once()
    notification = mock_notification.return_value
    notification.send.assert_called_once_with()


@patch("agents.providers.agent_prompt")
@patch("scrobbles.notifications.AgentSessionResponseNtfyNotification")
@patch("scrobbles.tasks.scrobble_agent_session_complete.apply_async")
def test_scrobble_agent_session_prompt_no_ntfy_on_error(
    mock_complete, mock_notification, mock_agent_prompt, user
):
    scrobble = _mk_scrobble(
        user,
        turns=[
            {
                "prompt_id": "abc-123",
                "prompt": "hello",
                "response": None,
            }
        ],
    )
    mock_agent_prompt.side_effect = ValueError("boom")

    scrobble_agent_session_prompt(scrobble.id, "abc-123")

    mock_notification.assert_not_called()


@patch("scrobbles.tasks.scrobble_agent_session_prompt.apply_async")
@patch("agents.providers.agent_prompt")
@patch("scrobbles.tasks.scrobble_agent_session_complete.apply_async")
def test_scrobble_agent_session_prompt_retries_transient_error(
    mock_complete, mock_agent_prompt, mock_retry, user, settings
):
    settings.CELERY_TASK_ALWAYS_EAGER = False
    settings.AGENT_PROVIDER_MAX_ATTEMPTS = 3
    settings.AGENT_PROVIDER_RETRY_BACKOFF_BASE = 60
    settings.AGENT_PROVIDER_RETRY_MAX_BACKOFF = 300
    scrobble = _mk_scrobble(
        user,
        turns=[
            {
                "prompt_id": "abc-123",
                "prompt": "hello",
                "response": None,
            }
        ],
    )
    mock_agent_prompt.side_effect = _http_error(429, {"Retry-After": "30"})

    scrobble_agent_session_prompt(scrobble.id, "abc-123")

    scrobble.refresh_from_db()
    turn = scrobble.log["turns"][0]
    assert turn["response"] is None
    assert turn["error"] is True
    assert turn["retryable"] is True
    assert turn["attempts"] == 1
    mock_retry.assert_called_once_with(args=[scrobble.id, "abc-123"], countdown=30)
    mock_complete.assert_not_called()


@patch("scrobbles.tasks.scrobble_agent_session_prompt.apply_async")
@patch("agents.providers.agent_prompt")
@patch("scrobbles.tasks.scrobble_agent_session_complete.apply_async")
def test_scrobble_agent_session_prompt_retry_backoff_without_retry_after(
    mock_complete, mock_agent_prompt, mock_retry, user, settings
):
    settings.CELERY_TASK_ALWAYS_EAGER = False
    settings.AGENT_PROVIDER_MAX_ATTEMPTS = 3
    settings.AGENT_PROVIDER_RETRY_BACKOFF_BASE = 60
    settings.AGENT_PROVIDER_RETRY_MAX_BACKOFF = 300
    scrobble = _mk_scrobble(
        user,
        turns=[
            {
                "prompt_id": "abc-123",
                "prompt": "hello",
                "response": None,
            }
        ],
    )
    mock_agent_prompt.side_effect = _http_error(503)

    scrobble_agent_session_prompt(scrobble.id, "abc-123")

    mock_retry.assert_called_once_with(args=[scrobble.id, "abc-123"], countdown=60)


@patch("scrobbles.tasks.scrobble_agent_session_prompt.apply_async")
@patch("agents.providers.agent_prompt")
@patch("scrobbles.tasks.scrobble_agent_session_complete.apply_async")
def test_scrobble_agent_session_prompt_gives_up_after_max_attempts(
    mock_complete, mock_agent_prompt, mock_retry, user, settings
):
    settings.CELERY_TASK_ALWAYS_EAGER = False
    settings.AGENT_PROVIDER_MAX_ATTEMPTS = 2
    scrobble = _mk_scrobble(
        user,
        turns=[
            {
                "prompt_id": "abc-123",
                "prompt": "hello",
                "response": None,
                "error": True,
                "retryable": True,
                "attempts": 1,
            }
        ],
    )
    mock_agent_prompt.side_effect = _http_error(429, {"Retry-After": "30"})

    scrobble_agent_session_prompt(scrobble.id, "abc-123")

    scrobble.refresh_from_db()
    turn = scrobble.log["turns"][0]
    assert turn["response"] == "Error: Provider returned HTTP 429 (Too Many Requests)"
    assert turn["error"] is True
    assert turn["retryable"] is False
    assert turn["attempts"] == 2
    mock_retry.assert_not_called()
    mock_complete.assert_called_once_with(
        args=[scrobble.id], countdown=AGENT_SESSION_AUTO_COMPLETE_SECONDS
    )


@patch("scrobbles.tasks.scrobble_agent_session_prompt.apply_async")
@patch("agents.providers.agent_prompt")
@patch("scrobbles.tasks.scrobble_agent_session_complete.apply_async")
def test_scrobble_agent_session_prompt_eager_mode_is_terminal(
    mock_complete, mock_agent_prompt, mock_retry, user, settings
):
    settings.CELERY_TASK_ALWAYS_EAGER = True
    scrobble = _mk_scrobble(
        user,
        turns=[
            {
                "prompt_id": "abc-123",
                "prompt": "hello",
                "response": None,
            }
        ],
    )
    mock_agent_prompt.side_effect = _http_error(429, {"Retry-After": "30"})

    scrobble_agent_session_prompt(scrobble.id, "abc-123")

    scrobble.refresh_from_db()
    turn = scrobble.log["turns"][0]
    assert turn["response"] == "Error: Provider returned HTTP 429 (Too Many Requests)"
    assert turn["retryable"] is False
    mock_retry.assert_not_called()


@patch("agents.providers.agent_prompt")
def test_scrobble_agent_session_prompt_skips_answered_turn(mock_agent_prompt, user):
    scrobble = _mk_scrobble(
        user,
        turns=[
            {
                "prompt_id": "abc-123",
                "prompt": "hello",
                "response": "hi back",
            }
        ],
    )

    scrobble_agent_session_prompt(scrobble.id, "abc-123")

    mock_agent_prompt.assert_not_called()
    scrobble.refresh_from_db()
    assert scrobble.log["turns"][0]["response"] == "hi back"


@patch("scrobbles.notifications.AgentSessionFailedNtfyNotification")
@patch("agents.providers.agent_prompt")
@patch("scrobbles.tasks.scrobble_agent_session_complete.apply_async")
def test_scrobble_agent_session_prompt_sends_ntfy_on_terminal_error(
    mock_complete, mock_agent_prompt, mock_notification, user
):
    scrobble = _mk_scrobble(
        user,
        turns=[
            {
                "prompt_id": "abc-123",
                "prompt": "hello",
                "response": None,
            }
        ],
    )
    mock_agent_prompt.side_effect = ValueError("boom")

    scrobble_agent_session_prompt(scrobble.id, "abc-123")

    mock_notification.assert_called_once()
    mock_notification.return_value.send.assert_called_once_with()
    mock_complete.assert_called_once_with(
        args=[scrobble.id], countdown=AGENT_SESSION_AUTO_COMPLETE_SECONDS
    )


@patch("agents.providers.agent_prompt")
@patch("scrobbles.tasks.scrobble_agent_session_complete.apply_async")
def test_scrobble_agent_session_prompt_success_clears_retry_state(
    mock_complete, mock_agent_prompt, user
):
    scrobble = _mk_scrobble(
        user,
        turns=[
            {
                "prompt_id": "abc-123",
                "prompt": "hello",
                "response": None,
                "error": True,
                "retryable": True,
                "attempts": 1,
            }
        ],
    )
    mock_agent_prompt.return_value = {
        "text": "hi back",
        "provider": settings.AGENT_PROVIDER,
        "model": settings.LLM_MODEL,
    }

    scrobble_agent_session_prompt(scrobble.id, "abc-123")

    scrobble.refresh_from_db()
    turn = scrobble.log["turns"][0]
    assert turn["response"] == "hi back"
    assert "error" not in turn
    assert "retryable" not in turn
    assert "attempts" not in turn


# --- auto-complete ---


def test_scrobble_agent_session_complete_marks_done(user):
    scrobble = _mk_scrobble(
        user,
        in_progress=True,
        turns=[{"prompt_id": "abc-123", "prompt": "hello", "response": "hi back"}],
    )

    scrobble_agent_session_complete(scrobble.id)

    scrobble.refresh_from_db()
    assert scrobble.in_progress is False
    assert scrobble.played_to_completion is True


@patch("scrobbles.tasks.scrobble_agent_session_complete.apply_async")
def test_scrobble_agent_session_complete_reschedules_when_pending(mock_complete, user):
    scrobble = _mk_scrobble(
        user,
        in_progress=True,
        turns=[{"prompt_id": "abc-123", "prompt": "hello", "response": None}],
    )

    scrobble_agent_session_complete(scrobble.id)

    scrobble.refresh_from_db()
    assert scrobble.in_progress is True
    mock_complete.assert_called_once_with(
        args=[scrobble.id], countdown=AGENT_SESSION_AUTO_COMPLETE_SECONDS
    )


def test_scrobble_agent_session_complete_unknown_scrobble(user):
    scrobble_agent_session_complete(999999)


# --- views ---


@patch("scrobbles.tasks.scrobble_agent_session_prompt.delay")
def test_manual_scrobble_view_routes_plain_text_to_agent(mock_delay, client, user):
    client.force_login(user)
    response = client.post(
        reverse("scrobbles:lookup-manual-scrobble"),
        {"item_id": "what is the meaning of life?"},
    )

    assert response.status_code == 302
    assert response.url == reverse("agents:agent_model_select")
    assert client.session["agent_prompt"] == "what is the meaning of life?"
    assert not Scrobble.objects.filter(user_id=user.id).exists()


@patch("agents.views.list_openrouter_free_models")
@patch("agents.views.list_gemini_agent_models")
def test_agent_model_select_view_lists_models(mock_gemini, mock_or, client, user):
    mock_gemini.return_value = [
        {"provider": "gemini", "id": "gemini-3.6-flash", "name": "gemini-3.6-flash"}
    ]
    mock_or.return_value = [
        {"provider": "openrouter", "id": "free/model-a", "name": "Model A"},
        {"provider": "openrouter", "id": "free/model-b", "name": "Model B"},
    ]
    client.force_login(user)
    session = client.session
    session["agent_prompt"] = "what is 2+2?"
    session.save()
    response = client.get(reverse("agents:agent_model_select"))

    assert response.status_code == 200
    assert b"what is 2+2?" in response.content
    assert b"free/model-a" in response.content
    assert b"Model A" in response.content
    assert b"gemini-3.6-flash" in response.content
    assert b"Google Gemini" in response.content
    expected = mock_gemini.return_value + mock_or.return_value
    assert client.session["agent_models"] == expected


@patch("agents.views.list_openrouter_free_models")
@patch("agents.views.list_gemini_agent_models")
def test_agent_model_select_view_handles_errors(mock_gemini, mock_or, client, user):
    mock_gemini.return_value = []
    mock_or.side_effect = ValueError("boom")
    client.force_login(user)
    session = client.session
    session["agent_prompt"] = "hi"
    session.save()
    response = client.get(reverse("agents:agent_model_select"))

    assert response.status_code == 200
    assert b"Could not load models from OpenRouter" in response.content


@patch("scrobbles.tasks.scrobble_agent_session_prompt.delay")
def test_agent_scrobble_from_model_creates_session(mock_delay, client, user):
    client.force_login(user)
    session = client.session
    session["agent_prompt"] = "what is 2+2?"
    session.save()
    response = client.post(
        reverse("agents:agent_scrobble_from_model"),
        {"model": "free/model-a", "prompt": "what is 2+2?"},
    )

    assert response.status_code == 302
    scrobble = Scrobble.objects.filter(user_id=user.id).first()
    assert scrobble is not None
    assert scrobble.agent_session.provider == "openrouter"
    assert scrobble.agent_session.model == "free/model-a"
    assert scrobble.log["title"] == "what is 2+2?"
    assert response.url == scrobble.get_absolute_url()
    assert "agent_prompt" not in client.session


@patch("scrobbles.tasks.scrobble_agent_session_prompt.delay")
def test_agent_scrobble_from_model_gemini_provider(mock_delay, client, user):
    client.force_login(user)
    session = client.session
    session["agent_prompt"] = "what is 2+2?"
    session.save()
    response = client.post(
        reverse("agents:agent_scrobble_from_model"),
        {"provider": "gemini", "model": "gemini-3.6-flash", "prompt": "what is 2+2?"},
    )

    assert response.status_code == 302
    scrobble = Scrobble.objects.filter(user_id=user.id).first()
    assert scrobble is not None
    assert scrobble.agent_session.provider == "gemini"
    assert scrobble.agent_session.model == "gemini-3.6-flash"
    assert response.url == scrobble.get_absolute_url()


@patch("scrobbles.tasks.scrobble_agent_session_prompt.delay")
def test_agent_scrobble_from_model_missing_model(mock_delay, client, user):
    client.force_login(user)
    response = client.post(reverse("agents:agent_scrobble_from_model"), {"model": ""})

    assert response.status_code == 302
    assert response.url == reverse("agents:agent_model_select")
    mock_delay.assert_not_called()
    assert not Scrobble.objects.filter(user_id=user.id).exists()


@patch("scrobbles.tasks.scrobble_agent_session_prompt.delay")
def test_manual_scrobble_agent_session_explicit_model(mock_delay, user):
    scrobble = manual_scrobble_agent_session(
        "hi", user.id, provider="openrouter", model="free/model-a"
    )

    assert scrobble.agent_session.provider == "openrouter"
    assert scrobble.agent_session.model == "free/model-a"
    assert scrobble.agent_session.title == "openrouter free/model-a"
    assert scrobble.log["turns"][0]["prompt"] == "hi"


@patch("scrobbles.tasks.scrobble_agent_session_prompt.delay")
def test_agent_session_partial_pending_polls(mock_delay, client, user):
    scrobble = _mk_scrobble(
        user,
        turns=[{"prompt_id": "abc-123", "prompt": "hello", "response": None}],
    )
    client.force_login(user)
    response = client.get(
        reverse("scrobbles:agent-session-partial", args=[scrobble.id])
    )

    assert response.status_code == 200
    assert b"every 5s" in response.content
    assert b"Thinking" in response.content


@patch("scrobbles.tasks.scrobble_agent_session_prompt.delay")
def test_agent_session_partial_completed_stops_polling(mock_delay, client, user):
    scrobble = _mk_scrobble(
        user,
        in_progress=False,
        turns=[
            {
                "prompt_id": "abc-123",
                "prompt": "hello",
                "response": "**hi back**",
            }
        ],
    )
    client.force_login(user)
    response = client.get(
        reverse("scrobbles:agent-session-partial", args=[scrobble.id])
    )

    assert response.status_code == 200
    assert b"every 5s" not in response.content
    assert b"hi back" in response.content


@patch("scrobbles.tasks.scrobble_agent_session_prompt.delay")
def test_agent_session_list_view(mock_delay, client, user):
    _mk_scrobble(user, in_progress=True)
    client.force_login(user)
    response = client.get(reverse("agents:agent_session_list"))

    assert response.status_code == 200
    assert settings.LLM_MODEL.encode() in response.content


@patch("scrobbles.tasks.scrobble_agent_session_prompt.delay")
def test_homepage_shows_agent_sessions_with_first_prompt_title(
    mock_delay, client, user
):
    first_prompt = "tell me about the history of cheese making in europe"
    _mk_scrobble(user, in_progress=False, title=first_prompt)
    client.force_login(user)
    response = client.get(reverse("vrobbler-home"))

    assert response.status_code == 200
    assert b"Agent Sessions" in response.content
    assert first_prompt.encode() in response.content
    assert settings.LLM_MODEL.encode() not in response.content


@patch("scrobbles.tasks.scrobble_agent_session_prompt.delay")
def test_agent_session_detail_view(mock_delay, client, user):
    scrobble = _mk_scrobble(user, in_progress=True)
    client.force_login(user)
    response = client.get(scrobble.media_obj.get_absolute_url())

    assert response.status_code == 200
    assert settings.LLM_MODEL.encode() in response.content


@patch("scrobbles.tasks.scrobble_agent_session_prompt.delay")
def test_manual_scrobble_agent_follow_up_appends_turn(mock_delay, user):
    scrobble = _mk_scrobble(
        user,
        in_progress=False,
        turns=[
            {
                "prompt_id": "abc-123",
                "prompt": "hello",
                "response": "hi back",
            }
        ],
    )

    result = manual_scrobble_agent_follow_up("tell me more", scrobble.id, user.id)

    assert result.id == scrobble.id
    scrobble.refresh_from_db()
    turns = scrobble.log["turns"]
    assert len(turns) == 2
    assert turns[1]["prompt"] == "tell me more"
    assert turns[1]["response"] is None
    assert scrobble.in_progress is True
    mock_delay.assert_called_once_with(scrobble.id, turns[1]["prompt_id"])


@patch("scrobbles.tasks.scrobble_agent_session_prompt.delay")
def test_manual_scrobble_agent_follow_up_unknown_scrobble(mock_delay, user):
    assert manual_scrobble_agent_follow_up("tell me more", 999999, user.id) is None
    mock_delay.assert_not_called()


@patch("scrobbles.tasks.scrobble_agent_session_prompt.delay")
def test_agent_session_followup_post_appends_turn(mock_delay, client, user):
    scrobble = _mk_scrobble(
        user,
        in_progress=False,
        turns=[
            {
                "prompt_id": "abc-123",
                "prompt": "hello",
                "response": "hi back",
            }
        ],
    )
    client.force_login(user)
    response = client.post(
        reverse("scrobbles:agent-session-followup", args=[scrobble.id]),
        {"prompt": "tell me more"},
    )

    assert response.status_code == 200
    assert b"tell me more" in response.content
    assert b"Thinking" in response.content
    assert b"every 5s" in response.content
    scrobble.refresh_from_db()
    assert len(scrobble.log["turns"]) == 2
    assert scrobble.in_progress is True


@patch("scrobbles.tasks.scrobble_agent_session_prompt.delay")
def test_agent_session_followup_post_empty_prompt(mock_delay, client, user):
    scrobble = _mk_scrobble(user, in_progress=False)
    client.force_login(user)
    response = client.post(
        reverse("scrobbles:agent-session-followup", args=[scrobble.id]),
        {"prompt": "   "},
    )

    assert response.status_code == 302
    mock_delay.assert_not_called()


@patch("scrobbles.tasks.scrobble_agent_session_prompt.delay")
def test_agent_session_followup_form_shown_when_complete(mock_delay, client, user):
    scrobble = _mk_scrobble(
        user,
        in_progress=False,
        turns=[
            {
                "prompt_id": "abc-123",
                "prompt": "hello",
                "response": "hi back",
            }
        ],
    )
    client.force_login(user)
    response = client.get(
        reverse("scrobbles:agent-session-partial", args=[scrobble.id])
    )

    assert response.status_code == 200
    assert b"Ask a follow-up" in response.content
    assert b"/follow-up/" in response.content
    assert b"csrfmiddlewaretoken" in response.content


@patch("scrobbles.tasks.scrobble_agent_session_prompt.delay")
def test_agent_session_retry_post_resets_and_dispatches(mock_delay, client, user):
    scrobble = _mk_scrobble(
        user,
        in_progress=False,
        turns=[
            {
                "prompt_id": "abc-123",
                "prompt": "hello",
                "response": "Error: boom",
                "error": True,
                "retryable": False,
                "attempts": 3,
            }
        ],
    )
    client.force_login(user)
    response = client.post(
        reverse("scrobbles:agent-session-retry", args=[scrobble.id]),
        {"prompt_id": "abc-123"},
    )

    assert response.status_code == 200
    assert b"Thinking" in response.content
    assert b"every 5s" in response.content
    scrobble.refresh_from_db()
    turn = scrobble.log["turns"][0]
    assert turn["response"] is None
    assert turn["error"] is False
    assert "retryable" not in turn
    assert "attempts" not in turn
    assert scrobble.in_progress is True
    mock_delay.assert_called_once_with(scrobble.id, "abc-123")


@patch("scrobbles.tasks.scrobble_agent_session_prompt.delay")
def test_agent_session_retry_post_skips_answered_turn(mock_delay, client, user):
    scrobble = _mk_scrobble(
        user,
        in_progress=False,
        turns=[
            {
                "prompt_id": "abc-123",
                "prompt": "hello",
                "response": "hi back",
            }
        ],
    )
    client.force_login(user)
    response = client.post(
        reverse("scrobbles:agent-session-retry", args=[scrobble.id]),
        {"prompt_id": "abc-123"},
    )

    assert response.status_code == 200
    mock_delay.assert_not_called()
    scrobble.refresh_from_db()
    assert scrobble.log["turns"][0]["response"] == "hi back"
    assert scrobble.in_progress is False


@patch("scrobbles.tasks.scrobble_agent_session_prompt.delay")
def test_agent_session_retry_post_skips_already_retrying(mock_delay, client, user):
    scrobble = _mk_scrobble(
        user,
        in_progress=True,
        turns=[
            {
                "prompt_id": "abc-123",
                "prompt": "hello",
                "response": None,
                "error": True,
                "retryable": True,
                "attempts": 1,
            }
        ],
    )
    client.force_login(user)
    response = client.post(
        reverse("scrobbles:agent-session-retry", args=[scrobble.id]),
        {"prompt_id": "abc-123"},
    )

    assert response.status_code == 200
    mock_delay.assert_not_called()
    scrobble.refresh_from_db()
    assert scrobble.log["turns"][0]["response"] is None
    assert scrobble.in_progress is True


# --- log data ---


def test_agent_session_logdata_pending():
    log = AgentSessionLogData(
        provider="gemini",
        model="gemini-2.5-flash",
        turns=[{"prompt_id": "1", "prompt": "hi", "response": None}],
    )
    assert log.pending is True
    log.turns[0]["response"] = "hi back"
    assert log.pending is False
    assert AgentSessionLogData(provider="gemini").pending is False


def test_agent_session_logdata_as_html():
    log = AgentSessionLogData(
        provider="gemini",
        model="gemini-2.5-flash",
        turns=[
            {"prompt_id": "1", "prompt": "**bold?**", "response": "yes"},
            {"prompt_id": "2", "prompt": "again", "response": None},
        ],
    )
    html = log.as_html()

    assert '<div class="agent-turn">' in html
    assert "Prompt 1" in html
    assert "<strong>bold?</strong>" in html
    assert "yes" in html
    assert 'id="response-1"' in html
    assert "Prompt 2" in html
    assert "Thinking" in html
    assert 'id="response-2"' in html


@patch("scrobbles.notifications.requests.post")
def test_agent_session_response_ntfy_links_to_response(mock_post, user):
    scrobble = _mk_scrobble(
        user,
        turns=[
            {
                "prompt_id": "abc-123",
                "prompt": "hello",
                "response": "hi back",
            }
        ],
        title="First prompt",
    )
    user.profile.ntfy_url = "https://ntfy.example.com/topic"
    user.profile.ntfy_enabled = True
    user.profile.save()

    scrobble = Scrobble.objects.get(pk=scrobble.pk)

    AgentSessionResponseNtfyNotification(scrobble, scrobble.log["turns"][0]).send()

    mock_post.assert_called_once()
    call = mock_post.call_args
    assert call.args[0] == "https://ntfy.example.com/topic"
    headers = call.kwargs["headers"]
    assert (
        headers["Click"]
        == f"https://example.com/scrobbles/{scrobble.pk}/#response-abc-123"
    )
    assert headers["Title"] == "Agent Response Ready"
    assert headers["Tags"] == "robot"
    body = call.kwargs["data"].decode("utf-8")
    assert "First prompt" in body
    assert "hi back" in body


def test_agent_session_logdata_as_html_retrying():
    log = AgentSessionLogData(
        provider="gemini",
        model="gemini-2.5-flash",
        turns=[
            {
                "prompt_id": "1",
                "prompt": "hi",
                "response": None,
                "error": True,
                "retryable": True,
                "attempts": 1,
            }
        ],
    )
    html = log.as_html()

    assert "Retrying" in html
    assert "attempt 1 of" in html
    assert "hx-post" not in html


def test_agent_session_logdata_as_html_retrying_attempts(settings):
    settings.AGENT_PROVIDER_MAX_ATTEMPTS = 5
    log = AgentSessionLogData(
        turns=[
            {
                "prompt_id": "1",
                "prompt": "hi",
                "response": None,
                "error": True,
                "retryable": True,
                "attempts": 3,
            }
        ],
    )
    assert "attempt 3 of 5" in log.as_html()


def test_agent_session_logdata_as_html_retry_button():
    log = AgentSessionLogData(
        turns=[
            {
                "prompt_id": "abc-123",
                "prompt": "hello",
                "response": "Error: boom",
                "error": True,
                "retryable": False,
            }
        ],
    )
    html = log.as_html(scrobble_id=42)

    assert 'hx-post="/agent-session/42/retry/"' in html
    assert 'value="abc-123"' in html
    assert "Retry" in html
    assert "hx-post" not in log.as_html()


def test_agent_session_logdata_as_html_escapes_prompt_id():
    log = AgentSessionLogData(
        turns=[
            {
                "prompt_id": '"><script>',
                "prompt": "hello",
                "response": "Error: boom",
                "error": True,
            }
        ],
    )
    html = log.as_html(scrobble_id=42)

    assert "&quot;&gt;&lt;script&gt;" in html
    assert "<script>" not in html


@patch("scrobbles.notifications.requests.post")
def test_agent_session_failed_ntfy_links_to_response(mock_post, user):
    scrobble = _mk_scrobble(
        user,
        turns=[
            {
                "prompt_id": "abc-123",
                "prompt": "hello",
                "response": "Error: boom",
                "error": True,
                "retryable": False,
            }
        ],
        title="First prompt",
    )
    user.profile.ntfy_url = "https://ntfy.example.com/topic"
    user.profile.ntfy_enabled = True
    user.profile.save()

    scrobble = Scrobble.objects.get(pk=scrobble.pk)

    AgentSessionFailedNtfyNotification(scrobble, scrobble.log["turns"][0]).send()

    mock_post.assert_called_once()
    call = mock_post.call_args
    assert call.args[0] == "https://ntfy.example.com/topic"
    headers = call.kwargs["headers"]
    assert (
        headers["Click"]
        == f"https://example.com/scrobbles/{scrobble.pk}/#response-abc-123"
    )
    assert headers["Title"] == "Agent Response Failed"
    assert headers["Tags"] == "warning"
    assert headers["Priority"] == "high"
    body = call.kwargs["data"].decode("utf-8")
    assert "First prompt" in body
    assert "Error: boom" in body