secstate/vrobbler
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[agents] Notify on agent responses and retry failed prompts

Browse Source
This commit is contained in:
Colin Powell
2026-08-03 10:25:39 -04:00
parent 2ce3b756e6
commit b8eb13083f
11 changed files with 856 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

567
tests/agents_tests/test_agent_sessions.py
View File

@ -2,22 +2,30 @@ 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,
@ -355,6 +363,63 @@ def test_list_gemini_agent_models_no_key(settings):
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 ---
@ -547,6 +612,265 @@ def test_scrobble_agent_session_prompt_stores_error(
)
@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 ---
@ -877,6 +1201,96 @@ def test_agent_session_followup_form_shown_when_complete(mock_delay, client, use
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 ---
@ -907,5 +1321,158 @@ def test_agent_session_logdata_as_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