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, groq_agent_prompt, is_transient_error, list_gemini_agent_models, list_groq_agent_models, list_mistral_agent_models, list_openrouter_free_models, mistral_agent_prompt, 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_agent_prompt_groq_dispatches(settings): settings.AGENT_PROVIDER = "groq" settings.GROQ_API_KEY = "gsk-test" with patch("agents.providers.groq_agent_prompt", return_value={"text": "x"}) as m: agent_prompt("hi", model="llama-3.3-70b-versatile") m.assert_called_once_with("hi", history=None, model="llama-3.3-70b-versatile") def test_agent_prompt_mistral_dispatches(settings): settings.AGENT_PROVIDER = "mistral" settings.MISTRAL_API_KEY = "msk-test" with patch( "agents.providers.mistral_agent_prompt", return_value={"text": "x"} ) as m: agent_prompt("hi", model="mistral-small-latest") m.assert_called_once_with("hi", history=None, model="mistral-small-latest") 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_groq_agent_prompt(settings): settings.GROQ_API_KEY = "gsk-test" settings.LLM_MODEL = "llama-3.3-70b-versatile" 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 = groq_agent_prompt("what is 2+2?") mock_post.assert_called_once() assert ( mock_post.call_args.args[0] == "https://api.groq.com/openai/v1/chat/completions" ) kwargs = mock_post.call_args.kwargs assert kwargs["headers"] == {"Authorization": "Bearer gsk-test"} assert kwargs["json"] == { "model": "llama-3.3-70b-versatile", "messages": [{"role": "user", "content": "what is 2+2?"}], } assert result == { "text": "the answer", "provider": "groq", "model": "llama-3.3-70b-versatile", } def test_groq_agent_prompt_missing_key(settings): settings.GROQ_API_KEY = "" with pytest.raises(ValueError, match="VROBBLER_GROQ_API_KEY is not set"): groq_agent_prompt("hi") def test_groq_agent_prompt_bad_response(settings): settings.GROQ_API_KEY = "gsk-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 Groq response"): groq_agent_prompt("hi") def test_groq_agent_prompt_with_history(settings): settings.GROQ_API_KEY = "gsk-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: groq_agent_prompt("times 3?", history=history, model="llama-3.3-70b-versatile") 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_mistral_agent_prompt(settings): settings.MISTRAL_API_KEY = "msk-test" settings.LLM_MODEL = "mistral-small-latest" 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 = mistral_agent_prompt("what is 2+2?") mock_post.assert_called_once() assert mock_post.call_args.args[0] == "https://api.mistral.ai/v1/chat/completions" kwargs = mock_post.call_args.kwargs assert kwargs["headers"] == {"Authorization": "Bearer msk-test"} assert kwargs["json"] == { "model": "mistral-small-latest", "messages": [{"role": "user", "content": "what is 2+2?"}], } assert result == { "text": "the answer", "provider": "mistral", "model": "mistral-small-latest", } def test_mistral_agent_prompt_missing_key(settings): settings.MISTRAL_API_KEY = "" with pytest.raises(ValueError, match="VROBBLER_MISTRAL_API_KEY is not set"): mistral_agent_prompt("hi") def test_mistral_agent_prompt_bad_response(settings): settings.MISTRAL_API_KEY = "msk-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 Mistral response"): mistral_agent_prompt("hi") def test_mistral_agent_prompt_with_history(settings): settings.MISTRAL_API_KEY = "msk-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: mistral_agent_prompt("times 3?", history=history, model="mistral-small-latest") 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() == [] def test_list_groq_agent_models(settings): settings.GROQ_API_KEY = "gsk-test" settings.GROQ_AGENT_MODELS = "llama-3.3-70b-versatile, openai/gpt-oss-20b" assert list_groq_agent_models() == [ { "provider": "groq", "id": "llama-3.3-70b-versatile", "name": "llama-3.3-70b-versatile", }, { "provider": "groq", "id": "openai/gpt-oss-20b", "name": "openai/gpt-oss-20b", }, ] def test_list_groq_agent_models_no_key(settings): settings.GROQ_API_KEY = "" assert list_groq_agent_models() == [] def test_list_mistral_agent_models(settings): settings.MISTRAL_API_KEY = "msk-test" settings.MISTRAL_AGENT_MODELS = "mistral-small-latest, codestral-latest" assert list_mistral_agent_models() == [ { "provider": "mistral", "id": "mistral-small-latest", "name": "mistral-small-latest", }, { "provider": "mistral", "id": "codestral-latest", "name": "codestral-latest", }, ] def test_list_mistral_agent_models_no_key(settings): settings.MISTRAL_API_KEY = "" assert list_mistral_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") @patch("agents.views.list_groq_agent_models") @patch("agents.views.list_mistral_agent_models") def test_agent_model_select_view_lists_models( mock_mistral, mock_groq, mock_gemini, mock_or, client, user ): mock_gemini.return_value = [ {"provider": "gemini", "id": "gemini-3.6-flash", "name": "gemini-3.6-flash"} ] mock_groq.return_value = [ { "provider": "groq", "id": "llama-3.3-70b-versatile", "name": "llama-3.3-70b-versatile", } ] mock_mistral.return_value = [ { "provider": "mistral", "id": "mistral-small-latest", "name": "mistral-small-latest", } ] 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 assert b"llama-3.3-70b-versatile" in response.content assert b"Groq" in response.content assert b"mistral-small-latest" in response.content assert b"Mistral" in response.content expected = ( mock_gemini.return_value + mock_or.return_value + mock_groq.return_value + mock_mistral.return_value ) assert client.session["agent_models"] == expected @patch("agents.views.list_openrouter_free_models") @patch("agents.views.list_gemini_agent_models") @patch("agents.views.list_groq_agent_models") @patch("agents.views.list_mistral_agent_models") def test_agent_model_select_view_handles_errors( mock_mistral, mock_groq, mock_gemini, mock_or, client, user ): mock_gemini.return_value = [] mock_groq.return_value = [] mock_mistral.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"AgentSession" 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_forks_completed_session(mock_delay, user): original = _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", original.id, user.id) assert result.id != original.id assert result.agent_session_id == original.agent_session_id assert result.in_progress is True assert result.log["title"] == original.log["title"] assert result.log["forked_from_id"] == original.id original.refresh_from_db() assert len(original.log["turns"]) == 1 assert original.in_progress is False turns = result.log["turns"] assert len(turns) == 2 assert turns[0] == original.log["turns"][0] assert turns[1]["prompt"] == "tell me more" assert turns[1]["response"] is None mock_delay.assert_called_once_with(result.id, turns[1]["prompt_id"]) @patch("scrobbles.tasks.scrobble_agent_session_prompt.delay") def test_manual_scrobble_agent_follow_up_appends_turn_when_in_progress( mock_delay, user ): scrobble = _mk_scrobble( user, in_progress=True, 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 assert "forked_from_id" not in scrobble.log 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_forks_completed_session(mock_delay, client, user): original = _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=[original.id]), {"prompt": "tell me more"}, ) assert response.status_code == 200 assert response["HX-Redirect"] == reverse( "scrobbles:detail", args=[response.context["object"].id] ) forked = response.context["object"] assert forked.id != original.id assert b"tell me more" in response.content assert b"Thinking" in response.content assert b"every 5s" in response.content original.refresh_from_db() assert len(original.log["turns"]) == 1 assert original.in_progress is False @patch("scrobbles.tasks.scrobble_agent_session_prompt.delay") def test_agent_session_followup_post_appends_turn_when_in_progress( mock_delay, client, user ): scrobble = _mk_scrobble( user, in_progress=True, 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 "HX-Redirect" not in response 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 '
' in html assert "Prompt 1" in html assert "bold?" 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": '">