diff --git a/backend/app/core/retry.py b/backend/app/core/retry.py new file mode 100644 index 000000000..292247775 --- /dev/null +++ b/backend/app/core/retry.py @@ -0,0 +1,11 @@ +import httpx +import requests + + +def is_transient(error: BaseException) -> bool: + """5xx from upstream, or a connection that never reached it.""" + if isinstance(error, (httpx.HTTPStatusError, requests.HTTPError)): + return error.response is not None and error.response.status_code >= 500 + return isinstance( + error, (httpx.ConnectError, httpx.ConnectTimeout, requests.ConnectionError) + ) diff --git a/backend/app/services/llm/guardrails.py b/backend/app/services/llm/guardrails.py index e84aa7d74..61002fed5 100644 --- a/backend/app/services/llm/guardrails.py +++ b/backend/app/services/llm/guardrails.py @@ -9,8 +9,10 @@ import httpx from fastapi import HTTPException from opentelemetry import trace +from tenacity import retry, retry_if_exception, stop_after_attempt, wait_exponential from app.core.config import settings +from app.core.retry import is_transient from app.models.llm.request import InlineValidator, InlineValidatorTypeEnum, Validator logger = logging.getLogger(__name__) @@ -91,7 +93,13 @@ def proxy_guardrails_request( ), httpx.Client(timeout=GUARDRAILS_PROXY_TIMEOUT_SECONDS) as client, ): - response = client.request( + # Status isn't raised, so only connect failures retry: safe for any method. + response = retry( + stop=stop_after_attempt(4), + wait=wait_exponential(max=10), + retry=retry_if_exception(is_transient), + reraise=True, + )(client.request)( method, url, params=query, json=json_body, headers=headers ) except httpx.RequestError as e: @@ -371,19 +379,29 @@ def run_guardrails_validation( ), httpx.Client(timeout=45.0) as client, ): - response = client.post( - url, - json=payload, - params={"suppress_pass_logs": str(suppress_pass_logs).lower()}, - headers=headers, - ) - elapsed_ms = int((time.monotonic() - started) * 1000) - logger.info( - f"[run_guardrails_validation] Response received | job_id={job_id}, " - f"status={response.status_code}, elapsed_ms={elapsed_ms}, " + + @retry( + stop=stop_after_attempt(4), + wait=wait_exponential(max=10), + retry=retry_if_exception(is_transient), + reraise=True, ) - response.raise_for_status() - return response.json() + def _post() -> dict[str, Any]: + response = client.post( + url, + json=payload, + params={"suppress_pass_logs": str(suppress_pass_logs).lower()}, + headers=headers, + ) + elapsed_ms = int((time.monotonic() - started) * 1000) + logger.info( + f"[run_guardrails_validation] Response received | job_id={job_id}, " + f"status={response.status_code}, elapsed_ms={elapsed_ms}, " + ) + response.raise_for_status() + return response.json() + + return _post() except Exception as e: elapsed_ms = int((time.monotonic() - started) * 1000) if _is_auth_error(e): @@ -458,6 +476,12 @@ def _build_params(validator_ids: list[UUID]) -> dict[str, Any]: httpx.Client(timeout=10.0) as client, ): + @retry( + stop=stop_after_attempt(4), + wait=wait_exponential(max=10), + retry=retry_if_exception(is_transient), + reraise=True, + ) def _fetch_by_ids(validator_ids: list[UUID]) -> list[dict[str, Any]]: if not validator_ids: return [] diff --git a/backend/app/services/llm/jobs.py b/backend/app/services/llm/jobs.py index 22a6dee95..d15b12191 100644 --- a/backend/app/services/llm/jobs.py +++ b/backend/app/services/llm/jobs.py @@ -16,10 +16,12 @@ from gevent import Timeout from opentelemetry import trace from sqlmodel import Session +from tenacity import retry, retry_if_exception, stop_after_attempt, wait_exponential from app.celery.utils import start_llm_chain_job, start_llm_job from app.core.db import engine from app.core.providers import Provider +from app.core.retry import is_transient from app.core.langfuse.langfuse import observe_llm_execution from app.core.telemetry import ( LogModuleEnum, @@ -859,6 +861,28 @@ def _run_output_guardrails( return result +# ReadTimeout isn't retried: the proxy may already have run (and billed) the call. +@retry( + stop=stop_after_attempt(4), + wait=wait_exponential(max=10), + retry=retry_if_exception(is_transient), + reraise=True, +) +def _post_to_proxy(url: str, api_key: str, input_text: str) -> dict: + with httpx.Client(timeout=60.0) as client: + resp = client.post( + url, + headers={ + "Authorization": f"Bearer {api_key}", + "Content-Type": "application/json", + "Accept": "application/json", + }, + json={"input": input_text}, + ) + resp.raise_for_status() + return resp.json() + + def _execute_proxy_call( session: Session, *, @@ -927,18 +951,9 @@ def _execute_proxy_call( try: with suppress_http_instrumentation(): - with httpx.Client(timeout=60.0) as client: - resp = client.post( - client_llm_url, - headers={ - "Authorization": f"Bearer {creds['api_key']}", - "Content-Type": "application/json", - "Accept": "application/json", - }, - json={"input": query.input.content.value}, - ) - resp.raise_for_status() - proxy_data = resp.json() + proxy_data = _post_to_proxy( + client_llm_url, creds["api_key"], query.input.content.value + ) except httpx.HTTPError as e: proxy_span.record_exception(e) proxy_span.set_status(trace.Status(trace.StatusCode.ERROR, str(e))) @@ -950,7 +965,8 @@ def _execute_proxy_call( return BlockResult( error=f"Proxy call failed: {str(e)}", llm_call_id=ctx.llm_call_id, - retryable=True, + # Transient failures were already retried; don't stack fast-eval retries. + retryable=not is_transient(e), ) try: diff --git a/backend/app/services/llm/providers/claude.py b/backend/app/services/llm/providers/claude.py index 5ae51461f..2832ae0ad 100644 --- a/backend/app/services/llm/providers/claude.py +++ b/backend/app/services/llm/providers/claude.py @@ -99,7 +99,7 @@ def __init__(self, client: Anthropic): def create_client(credentials: dict[str, Any]) -> Any: if "api_key" not in credentials: raise ValueError("Anthropic credentials not configured for this project.") - return Anthropic(api_key=credentials["api_key"]) + return Anthropic(api_key=credentials["api_key"], max_retries=3) @staticmethod def format_parts( diff --git a/backend/app/services/llm/providers/eleven_ai.py b/backend/app/services/llm/providers/eleven_ai.py index c6ac4cc96..fc1a8a81c 100644 --- a/backend/app/services/llm/providers/eleven_ai.py +++ b/backend/app/services/llm/providers/eleven_ai.py @@ -10,9 +10,11 @@ SpeechToTextConvertResponse, ) from elevenlabs.core.api_error import ApiError as ElevenLabsApiError +from tenacity import retry, retry_if_exception, stop_after_attempt, wait_exponential # from elevenlabs.types import SpeechToTextConvertResponse from app.core.audio_utils import AudioRef +from app.core.retry import is_transient from app.models.llm import ( @@ -36,6 +38,17 @@ logger = logging.getLogger(__name__) +# The SDK's own `max_retries` is unusable: its counter starts at 2, and streaming TTS never retries. +_retry_transient = retry( + stop=stop_after_attempt(4), + wait=wait_exponential(max=10), + retry=retry_if_exception( + lambda e: is_transient(e) + or (isinstance(e, ElevenLabsApiError) and (e.status_code or 0) >= 500) + ), + reraise=True, +) + class ElevenlabsAIProvider(BaseProvider): def __init__(self, client: ElevenLabs): @@ -132,11 +145,9 @@ def _execute_stt( with resolved_input.to_path() as parsed_input_path, open( parsed_input_path, "rb" ) as audio_file: - elevenlabs_response: SpeechToTextConvertResponse = ( - self.client.speech_to_text.convert( - file=audio_file, model_id=model_id, **stt_kwargs - ) - ) + elevenlabs_response: SpeechToTextConvertResponse = _retry_transient( + self.client.speech_to_text.convert + )(file=audio_file, model_id=model_id, **stt_kwargs) if not elevenlabs_response.text: error_message = ( @@ -248,17 +259,21 @@ def _execute_tts( if voice_settings: tts_kwargs["voice_settings"] = voice_settings - # Call Elevenlabs TTS API - audio_iterator = self.client.text_to_speech.convert( - voice_id=voice_id, - text=parsed_text, - model_id=model_id, - output_format=output_format, - **tts_kwargs, - ) + # convert() streams: the request is only sent while the iterator is consumed, + # so the retry must wrap the join too. + @_retry_transient + def _synthesize() -> bytes: + return b"".join( + self.client.text_to_speech.convert( + voice_id=voice_id, + text=parsed_text, + model_id=model_id, + output_format=output_format, + **tts_kwargs, + ) + ) - # Elevenlabs returns an iterator of audio bytes; collect and base64-encode - audio_bytes = b"".join(audio_iterator) + audio_bytes = _synthesize() if not audio_bytes: error_message = ( "[ELEVENLABS] TTS response contains no audio data. " diff --git a/backend/app/services/llm/providers/open_ai.py b/backend/app/services/llm/providers/open_ai.py index 13858a0b6..83ea70fae 100644 --- a/backend/app/services/llm/providers/open_ai.py +++ b/backend/app/services/llm/providers/open_ai.py @@ -36,7 +36,7 @@ def __init__(self, client: OpenAI): def create_client(credentials: dict[str, Any]) -> Any: if "api_key" not in credentials: raise ValueError("OpenAI credentials not configured for this project.") - return OpenAI(api_key=credentials["api_key"]) + return OpenAI(api_key=credentials["api_key"], max_retries=3) @staticmethod def format_parts( diff --git a/backend/app/services/llm/providers/sarvam_ai.py b/backend/app/services/llm/providers/sarvam_ai.py index e218d4931..164318282 100644 --- a/backend/app/services/llm/providers/sarvam_ai.py +++ b/backend/app/services/llm/providers/sarvam_ai.py @@ -115,7 +115,9 @@ def _execute_stt( with open(parsed_input_path, "rb") as audio_file: stt_kwargs["file"] = audio_file - sarvam_response = self.client.speech_to_text.transcribe(**stt_kwargs) + sarvam_response = self.client.speech_to_text.transcribe( + **stt_kwargs, request_options={"max_retries": 3} + ) if not sarvam_response.transcript: error_message = ( @@ -225,7 +227,9 @@ def _execute_tts( tts_kwargs["output_audio_codec"] = output_audio_codec # Call SarvamAI TTS with mapped parameters - sarvam_response = self.client.text_to_speech.convert(**tts_kwargs) + sarvam_response = self.client.text_to_speech.convert( + **tts_kwargs, request_options={"max_retries": 3} + ) # SarvamAI returns a list of base64-encoded audio strings # For single text input, take the first audio diff --git a/backend/app/tests/conftest.py b/backend/app/tests/conftest.py index 4a4ce9f41..173e3c13d 100644 --- a/backend/app/tests/conftest.py +++ b/backend/app/tests/conftest.py @@ -88,6 +88,12 @@ def aws_credentials(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setenv("AWS_DEFAULT_REGION", settings.AWS_DEFAULT_REGION) +@pytest.fixture +def no_retry_backoff(monkeypatch: pytest.MonkeyPatch) -> None: + """Tenacity binds nap.sleep at import (incl. nested decorators), so patch time.sleep.""" + monkeypatch.setattr("tenacity.nap.time.sleep", lambda _seconds: None) + + @pytest.fixture(scope="function") def client(db: Session) -> Generator[TestClient, None, None]: app.dependency_overrides[get_db] = lambda: db diff --git a/backend/app/tests/core/test_callback_ssrf.py b/backend/app/tests/core/test_callback_ssrf.py index 15dde46a0..48c654426 100644 --- a/backend/app/tests/core/test_callback_ssrf.py +++ b/backend/app/tests/core/test_callback_ssrf.py @@ -1,3 +1,4 @@ +import itertools import json import socket from typing import Any @@ -16,6 +17,8 @@ sign_webhook_payload, ) +pytestmark = pytest.mark.usefixtures("no_retry_backoff") + class TestIsPrivateIP: """Test suite for _is_private_ip function.""" @@ -237,7 +240,7 @@ def test_successful_callback( """Test successful callback execution.""" mock_session = MagicMock() mock_response = MagicMock() - mock_response.raise_for_status.return_value = None + mock_response.status_code = 200 mock_response.iter_content.return_value = [b"test"] mock_session.post.return_value = mock_response mock_session_class.return_value.__enter__.return_value = mock_session @@ -272,7 +275,7 @@ def test_callback_http_error( """Test that callback returns False on HTTP errors.""" mock_session = MagicMock() mock_response = MagicMock() - mock_response.raise_for_status.side_effect = requests.HTTPError("404 Not Found") + mock_response.status_code = 404 mock_session.post.return_value = mock_response mock_session_class.return_value.__enter__.return_value = mock_session @@ -288,7 +291,7 @@ def test_callback_disables_redirects( """Test that redirects are disabled to prevent redirect-based SSRF.""" mock_session = MagicMock() mock_response = MagicMock() - mock_response.raise_for_status.return_value = None + mock_response.status_code = 200 mock_response.iter_content.return_value = [b"test"] mock_session.post.return_value = mock_response mock_session_class.return_value.__enter__.return_value = mock_session @@ -306,7 +309,7 @@ def test_callback_uses_timeout( """Test that callback uses configured timeouts.""" mock_session = MagicMock() mock_response = MagicMock() - mock_response.raise_for_status.return_value = None + mock_response.status_code = 200 mock_response.iter_content.return_value = [b"test"] mock_session.post.return_value = mock_response mock_session_class.return_value.__enter__.return_value = mock_session @@ -326,7 +329,7 @@ def test_callback_sends_json_data( """Test that callback sends data as JSON.""" mock_session = MagicMock() mock_response = MagicMock() - mock_response.raise_for_status.return_value = None + mock_response.status_code = 200 mock_response.iter_content.return_value = [b"test"] mock_session.post.return_value = mock_response mock_session_class.return_value.__enter__.return_value = mock_session @@ -378,3 +381,57 @@ def test_signature_matches_manual_hmac(self): sig, returned_ts = sign_webhook_payload(secret, body, timestamp_ms=ts) assert sig == expected assert returned_ts == ts + + +def _callback_response(status_code: int) -> requests.Response: + response = requests.Response() + response.status_code = status_code + return response + + +@patch("app.utils.validate_callback_url") +class TestSendCallbackRetry: + URL = "https://api.example.com/callback" + + def test_recovers_after_transient_failures(self, _validate: Any) -> None: + # 302 counts as a failure: redirects are off, so nothing was delivered. + outcomes = [ + _callback_response(302), + requests.ConnectionError("reset"), + _callback_response(200), + ] + with patch.object(requests.Session, "post", side_effect=outcomes) as post: + assert send_callback(self.URL, {"status": "done"}) is True + assert post.call_count == 3 + + @pytest.mark.parametrize("status_code", [503, 404]) + def test_gives_up_after_four_attempts( + self, _validate: Any, status_code: int + ) -> None: + with patch.object( + requests.Session, "post", return_value=_callback_response(status_code) + ) as post: + assert send_callback(self.URL, {"status": "done"}) is False + assert post.call_count == 4 + + def test_each_attempt_is_signed_with_a_fresh_timestamp( + self, _validate: Any + ) -> None: + with ( + patch("app.utils.time.time", side_effect=itertools.count(1_700_000_000)), + patch.object( + requests.Session, "post", return_value=_callback_response(503) + ) as post, + ): + send_callback(self.URL, {"status": "done"}, webhook_secret="s3cret") + + timestamps = { + c.kwargs["headers"]["X-Webhook-Timestamp"] for c in post.mock_calls + } + assert len(timestamps) == 4 + + def test_invalid_url_is_never_posted(self, validate: Any) -> None: + validate.side_effect = ValueError("private IP") + with patch.object(requests.Session, "post") as post: + assert send_callback(self.URL, {"status": "done"}) is False + post.assert_not_called() diff --git a/backend/app/tests/core/test_retry.py b/backend/app/tests/core/test_retry.py new file mode 100644 index 000000000..fbb4c62f4 --- /dev/null +++ b/backend/app/tests/core/test_retry.py @@ -0,0 +1,27 @@ +import httpx +import pytest + +from app.core.retry import is_transient + +_REQUEST = httpx.Request("POST", "https://upstream.example/v1") + + +def _status_error(code: int) -> httpx.HTTPStatusError: + return httpx.HTTPStatusError( + "upstream", request=_REQUEST, response=httpx.Response(code, request=_REQUEST) + ) + + +@pytest.mark.parametrize( + "error, expected", + [ + (_status_error(500), True), + (_status_error(404), False), + (httpx.ConnectError("refused"), True), + (httpx.ConnectTimeout("slow handshake"), True), + # The request may have reached upstream, so a retry could double-execute. + (httpx.ReadTimeout("no response"), False), + ], +) +def test_is_transient(error: BaseException, expected: bool) -> None: + assert is_transient(error) is expected diff --git a/backend/app/tests/services/llm/providers/test_claude.py b/backend/app/tests/services/llm/providers/test_claude.py index 0758f97a9..bef8d1d1c 100644 --- a/backend/app/tests/services/llm/providers/test_claude.py +++ b/backend/app/tests/services/llm/providers/test_claude.py @@ -104,7 +104,7 @@ def test_returns_anthropic_client(self): with patch("app.services.llm.providers.claude.Anthropic") as mock_anthropic: mock_anthropic.return_value = MagicMock(name="anthropic-client") client = ClaudeProvider.create_client({"api_key": "sk-test"}) - mock_anthropic.assert_called_once_with(api_key="sk-test") + mock_anthropic.assert_called_once_with(api_key="sk-test", max_retries=3) assert client is mock_anthropic.return_value diff --git a/backend/app/tests/services/llm/providers/test_eleven_ai.py b/backend/app/tests/services/llm/providers/test_eleven_ai.py index 8ad4eaedd..e78e5ff06 100644 --- a/backend/app/tests/services/llm/providers/test_eleven_ai.py +++ b/backend/app/tests/services/llm/providers/test_eleven_ai.py @@ -12,8 +12,12 @@ QueryParams, ) from app.models.llm.constants import CompletionType +from elevenlabs.core.api_error import ApiError + from app.services.llm.providers.eleven_ai import ElevenlabsAIProvider +pytestmark = pytest.mark.usefixtures("no_retry_backoff") + def mock_elevenlabs_stt_response( text: str = "Hello world", @@ -203,6 +207,33 @@ def test_stt_include_provider_raw_response( assert result.provider_raw_response["text"] == "Test transcript" assert result.provider_raw_response["transcription_id"] == "txn_stt_123" + def test_stt_retries_5xx_then_succeeds( + self, provider, mock_client, stt_config, query_params, temp_audio_file + ): + mock_client.speech_to_text.convert.side_effect = [ + ApiError(status_code=503, body="down"), + mock_elevenlabs_stt_response(text="Hello"), + ] + + result, error = provider.execute(stt_config, query_params, temp_audio_file) + + assert error is None + assert result.response.output.content.value == "Hello" + assert mock_client.speech_to_text.convert.call_count == 2 + + def test_stt_does_not_retry_4xx( + self, provider, mock_client, stt_config, query_params, temp_audio_file + ): + mock_client.speech_to_text.convert.side_effect = ApiError( + status_code=400, body="bad" + ) + + result, error = provider.execute(stt_config, query_params, temp_audio_file) + + assert result is None + assert error is not None + assert mock_client.speech_to_text.convert.call_count == 1 + class TestElevenlabsProviderTTS: """Test cases for ElevenlabsAIProvider TTS functionality.""" @@ -451,6 +482,37 @@ def test_tts_usage_estimates(self, provider, mock_client, tts_config, query_para assert result.usage.output_tokens == 0 assert result.usage.total_tokens == 5 + def test_tts_retries_5xx_raised_while_streaming( + self, provider, mock_client, tts_config, query_params + ): + def failing_stream(): + raise ApiError(status_code=503, body="down") + yield b"" + + mock_client.text_to_speech.convert.side_effect = [ + failing_stream(), + iter([b"audio"]), + ] + + result, error = provider.execute(tts_config, query_params, "Namaste") + + assert error is None + assert result is not None + assert mock_client.text_to_speech.convert.call_count == 2 + + def test_tts_gives_up_after_four_attempts( + self, provider, mock_client, tts_config, query_params + ): + mock_client.text_to_speech.convert.side_effect = ApiError( + status_code=502, body="bad gateway" + ) + + result, error = provider.execute(tts_config, query_params, "Namaste") + + assert result is None + assert error is not None + assert mock_client.text_to_speech.convert.call_count == 4 + class TestElevenlabsProviderClientCreation: """Test cases for ElevenlabsAIProvider client creation.""" diff --git a/backend/app/tests/services/llm/providers/test_open_ai.py b/backend/app/tests/services/llm/providers/test_open_ai.py index 70d5ee16f..1afd28f1d 100644 --- a/backend/app/tests/services/llm/providers/test_open_ai.py +++ b/backend/app/tests/services/llm/providers/test_open_ai.py @@ -3,7 +3,7 @@ """ import pytest -from unittest.mock import MagicMock +from unittest.mock import MagicMock, patch import openai @@ -379,3 +379,9 @@ def test_execute_with_conversation_parameter_removed_when_no_config( # Verify old conversation was removed call_args = mock_client.responses.create.call_args assert "conversation" not in call_args[1] + + +def test_create_client_retries_thrice() -> None: + with patch("app.services.llm.providers.open_ai.OpenAI") as mock_openai: + OpenAIProvider.create_client({"api_key": "sk-test"}) + mock_openai.assert_called_once_with(api_key="sk-test", max_retries=3) diff --git a/backend/app/tests/services/llm/providers/test_sarvam_ai.py b/backend/app/tests/services/llm/providers/test_sarvam_ai.py index e73b2d481..61c2be046 100644 --- a/backend/app/tests/services/llm/providers/test_sarvam_ai.py +++ b/backend/app/tests/services/llm/providers/test_sarvam_ai.py @@ -132,6 +132,7 @@ def test_stt_success_with_translate_mode( # Verify translate mode was passed to API call_args = mock_client.speech_to_text.transcribe.call_args assert call_args.kwargs["mode"] == "translate" + assert call_args.kwargs["request_options"] == {"max_retries": 3} def test_stt_success_with_unknown_language( self, provider, mock_client, query_params, temp_audio_file @@ -301,6 +302,7 @@ def test_tts_with_mp3_codec(self, provider, mock_client, query_params): assert result.response.output.content.mime_type == "audio/mp3" call_args = mock_client.text_to_speech.convert.call_args assert call_args.kwargs["output_audio_codec"] == "mp3" + assert call_args.kwargs["request_options"] == {"max_retries": 3} def test_tts_with_ogg_codec(self, provider, mock_client, query_params): """Test TTS with OGG codec.""" diff --git a/backend/app/tests/services/llm/test_execute_llm_call.py b/backend/app/tests/services/llm/test_execute_llm_call.py index 0c913848b..cd97f43ad 100644 --- a/backend/app/tests/services/llm/test_execute_llm_call.py +++ b/backend/app/tests/services/llm/test_execute_llm_call.py @@ -23,6 +23,8 @@ from app.services.llm.jobs import execute_llm_call from app.tests.utils.utils import get_project +pytestmark = pytest.mark.usefixtures("no_retry_backoff") + VALIDATOR_CONFIG_ID = "00000000-0000-0000-0000-000000000001" PROXY_URL = "https://api.tap.example/v1/predictions" @@ -381,13 +383,22 @@ def test_provider_failure_is_retryable(self, db: Session, provider): assert result.retryable is True - def test_proxy_transport_failure_is_retryable( + def test_proxy_connect_failure_is_not_retryable_after_inline_retries( self, db: Session, provider, proxy_http ): proxy_http.post.side_effect = httpx.ConnectError("connection refused") result = call(db, build_blob(PROXY_COMPLETION)) + assert result.error.startswith("Proxy call failed:") + assert result.retryable is False + + def test_proxy_4xx_is_retryable(self, db: Session, provider, proxy_http): + request = httpx.Request("POST", PROXY_URL) + proxy_http.post.return_value = httpx.Response(400, request=request) + + result = call(db, build_blob(PROXY_COMPLETION)) + assert result.error.startswith("Proxy call failed:") assert result.retryable is True diff --git a/backend/app/tests/services/llm/test_guardrails.py b/backend/app/tests/services/llm/test_guardrails.py index 2b0b7f07f..433ea39e6 100644 --- a/backend/app/tests/services/llm/test_guardrails.py +++ b/backend/app/tests/services/llm/test_guardrails.py @@ -5,6 +5,7 @@ import httpx import pytest +from fastapi import HTTPException from sqlmodel import Session, select from app.core.config import settings @@ -24,12 +25,15 @@ GuardrailsOutcome, apply_guardrails, list_validators_config, + proxy_guardrails_request, run_guardrails_validation, summarize_validator_results, ) from app.tests.utils.utils import get_project +pytestmark = pytest.mark.usefixtures("no_retry_backoff") + TEST_JOB_ID = uuid.uuid4() TEST_TEXT = "hello world" TEST_CONFIG = [{"type": "pii_remover"}] @@ -740,3 +744,92 @@ def test_bypassed_is_not_blocked(self) -> None: assert outcome.bypassed is True assert outcome.blocked is False + + +def _guardrails_response(code: int, body: Any = None) -> httpx.Response: + request = httpx.Request("GET", f"{settings.KAAPI_GUARDRAILS_URL}/x") + return httpx.Response(code, json=body, request=request) + + +class TestGuardrailsRetry: + def _validate(self, *responses: httpx.Response) -> tuple[dict[str, Any], int]: + with patch.object(httpx.Client, "post", side_effect=responses) as post: + result = run_guardrails_validation( + TEST_TEXT, + TEST_CONFIG, + TEST_JOB_ID, + TEST_PROJECT_ID, + TEST_ORGANIZATION_ID, + ) + return result, post.call_count + + def test_validation_recovers_from_503(self) -> None: + result, calls = self._validate( + _guardrails_response(503), _guardrails_response(200, {"success": True}) + ) + + assert result == {"success": True} + assert calls == 2 + + def test_validation_bypasses_after_exhausting_503s(self) -> None: + result, calls = self._validate(*[_guardrails_response(503)] * 4) + + assert result["bypassed"] is True + assert calls == 4 + + def test_validation_auth_error_is_not_retried(self) -> None: + result, calls = self._validate( + _guardrails_response(401), _guardrails_response(200, {"success": True}) + ) + + assert result["auth_error"] is True + assert calls == 1 + + def test_list_validators_config_retries_connect_error(self) -> None: + validator = {"type": "uli_slur_match", "stage": "input"} + outcomes = [ + httpx.ConnectError("refused"), + _guardrails_response(200, {"success": True, "data": [validator]}), + ] + with patch.object(httpx.Client, "get", side_effect=outcomes): + input_guardrails, _ = list_validators_config( + input_validator_configs=[Validator(validator_config_id=uuid4())], + output_validator_configs=[], + organization_id=TEST_ORGANIZATION_ID, + project_id=TEST_PROJECT_ID, + ) + + assert input_guardrails == [validator] + + def test_proxy_connect_failure_retries_then_502(self) -> None: + with patch.object( + httpx.Client, "request", side_effect=httpx.ConnectError("refused") + ) as request: + with pytest.raises(HTTPException) as exc: + proxy_guardrails_request( + "POST", + "/validators/configs/", + organization_id=TEST_ORGANIZATION_ID, + project_id=TEST_PROJECT_ID, + json_body={"type": "pii_remover"}, + ) + + assert exc.value.status_code == 502 + assert request.call_count == 4 + + def test_proxy_upstream_5xx_is_passed_through_without_retry(self) -> None: + with patch.object( + httpx.Client, + "request", + return_value=_guardrails_response(503, {"detail": "down"}), + ) as request: + status, body = proxy_guardrails_request( + "GET", + "/validators/configs/", + organization_id=TEST_ORGANIZATION_ID, + project_id=TEST_PROJECT_ID, + ) + + assert status == 503 + assert body == {"detail": "down"} + assert request.call_count == 1 diff --git a/backend/app/tests/services/llm/test_jobs.py b/backend/app/tests/services/llm/test_jobs.py index ac41e2f7e..f6b4b8185 100644 --- a/backend/app/tests/services/llm/test_jobs.py +++ b/backend/app/tests/services/llm/test_jobs.py @@ -1,3 +1,4 @@ +import httpx import pytest from unittest.mock import patch, MagicMock from uuid import UUID, uuid4 @@ -36,10 +37,13 @@ execute_job, execute_chain_job, resolve_config_blob, + _post_to_proxy, ) from app.tests.utils.utils import get_project from app.tests.utils.test_data import create_test_config +pytestmark = pytest.mark.usefixtures("no_retry_backoff") + VALIDATOR_CONFIG_ID_1 = "00000000-0000-0000-0000-000000000001" VALIDATOR_CONFIG_ID_2 = "00000000-0000-0000-0000-000000000002" @@ -2997,3 +3001,41 @@ def test_resolve_both_native_and_kaapi_configs(self, db: Session): assert error_kaapi is None assert isinstance(resolved_kaapi.completion, KaapiTextCompletionConfig) assert resolved_kaapi.completion.provider == "openai" + + +class TestPostToProxyRetry: + URL = "https://proxy.example/v1/predict" + + def _post(self, *outcomes: int | Exception) -> tuple[dict | Exception, int]: + request = httpx.Request("POST", self.URL) + side_effect = [ + httpx.Response(o, json={"ok": o}, request=request) + if isinstance(o, int) + else o + for o in outcomes + ] + with patch.object(httpx.Client, "post", side_effect=side_effect) as post: + try: + result: dict | Exception = _post_to_proxy(self.URL, "key", "hi") + except httpx.HTTPError as e: + result = e + return result, post.call_count + + def test_recovers_from_connect_error_and_5xx(self) -> None: + result, calls = self._post(httpx.ConnectError("refused"), 502, 200) + + assert result == {"ok": 200} + assert calls == 3 + + def test_read_timeout_is_not_retried(self) -> None: + result, calls = self._post(httpx.ReadTimeout("slow"), 200) + + assert isinstance(result, httpx.ReadTimeout) + assert calls == 1 + + def test_4xx_is_not_retried(self) -> None: + result, calls = self._post(400, 200) + + assert isinstance(result, httpx.HTTPStatusError) + assert result.response.status_code == 400 + assert calls == 1 diff --git a/backend/app/tests/test_utils.py b/backend/app/tests/test_utils.py index 16d666d7d..976cad1e7 100644 --- a/backend/app/tests/test_utils.py +++ b/backend/app/tests/test_utils.py @@ -396,9 +396,7 @@ class TestSendCallbackWithSigning: @patch("requests.Session") def test_includes_signature_headers(self, mock_session_cls, mock_validate) -> None: mock_session = MagicMock() - mock_resp = MagicMock() - mock_resp.raise_for_status.return_value = None - mock_session.post.return_value = mock_resp + mock_session.post.return_value = MagicMock(status_code=200) mock_session_cls.return_value.__enter__.return_value = mock_session result = send_callback( @@ -419,6 +417,7 @@ def test_returns_false_for_blocked_url(self, mock_validate) -> None: assert result is False +@pytest.mark.usefixtures("no_retry_backoff") class TestSendCallbackLogging: _payload = { "success": True, @@ -433,10 +432,7 @@ def test_failure_logs_status_and_payload_without_user_data( self, mock_session_cls, mock_validate, caplog: pytest.LogCaptureFixture ) -> None: mock_session = MagicMock() - error_response = MagicMock(status_code=502) - mock_session.post.return_value.raise_for_status.side_effect = ( - requests.HTTPError("502 Bad Gateway", response=error_response) - ) + mock_session.post.return_value = MagicMock(status_code=502) mock_session_cls.return_value.__enter__.return_value = mock_session with caplog.at_level("ERROR", logger="app.utils"): diff --git a/backend/app/utils.py b/backend/app/utils.py index a01a342d3..1d215fc9f 100644 --- a/backend/app/utils.py +++ b/backend/app/utils.py @@ -28,6 +28,12 @@ from openai import OpenAI from pydantic import BaseModel, JsonValue from sqlmodel import Session +from tenacity import ( + retry, + retry_if_exception_type, + stop_after_attempt, + wait_exponential, +) from app.core import security from app.core.audio_utils import AudioRef @@ -579,7 +585,7 @@ def send_callback( attach X-Webhook-Signature / X-Webhook-Timestamp headers. Returns: - bool: True if callback succeeded, False otherwise + bool: True if callback succeeded, False once retries are exhausted """ try: validate_callback_url(str(callback_url)) @@ -588,37 +594,17 @@ def send_callback( return False log_payload = {k: v for k, v in data.items() if k != CALLBACK_USER_DATA_KEY} raw_body = json.dumps(data, separators=(",", ":")).encode() + span = trace.get_current_span() + span.set_attribute("callback.request.body.size", len(raw_body)) started_at = time.perf_counter() try: - headers = {"Content-Type": "application/json"} - - if webhook_secret: - signature, timestamp_ms = sign_webhook_payload(webhook_secret, raw_body) - headers["X-Webhook-Signature"] = signature - headers["X-Webhook-Timestamp"] = str(timestamp_ms) - span = trace.get_current_span() - span.set_attribute("callback.request.body.size", len(raw_body)) - with requests.Session() as session: - session.trust_env = False # Ignores environment proxies and other implicit settings for SSRF safety - - response = session.post( - callback_url, - data=raw_body, - headers=headers, - timeout=( - settings.CALLBACK_CONNECT_TIMEOUT, - settings.CALLBACK_READ_TIMEOUT, - ), - allow_redirects=False, - ) - span.set_attribute( - "callback.duration_ms", - round((time.perf_counter() - started_at) * 1000, 2), - ) - span.set_attribute("callback.response.status_code", response.status_code) - - response.raise_for_status() - return True + response = _post_callback(callback_url, raw_body, webhook_secret) + span.set_attribute( + "callback.duration_ms", + round((time.perf_counter() - started_at) * 1000, 2), + ) + span.set_attribute("callback.response.status_code", response.status_code) + return True except requests.RequestException as e: status = e.response.status_code if e.response is not None else None @@ -631,6 +617,41 @@ def send_callback( return False +@retry( + stop=stop_after_attempt(4), + wait=wait_exponential(max=10), + retry=retry_if_exception_type(requests.RequestException), + reraise=True, +) +def _post_callback( + callback_url: str, raw_body: bytes, webhook_secret: str | None +) -> requests.Response: + headers = {"Content-Type": "application/json"} + # Signed per attempt so the timestamp stays inside receivers' replay window. + if webhook_secret: + signature, timestamp_ms = sign_webhook_payload(webhook_secret, raw_body) + headers["X-Webhook-Signature"] = signature + headers["X-Webhook-Timestamp"] = str(timestamp_ms) + with requests.Session() as session: + session.trust_env = False # Ignores environment proxies and other implicit settings for SSRF safety + response = session.post( + callback_url, + data=raw_body, + headers=headers, + timeout=( + settings.CALLBACK_CONNECT_TIMEOUT, + settings.CALLBACK_READ_TIMEOUT, + ), + allow_redirects=False, + ) + # Not raise_for_status(): it passes 3xx, and with redirects off nothing was delivered. + if not 200 <= response.status_code < 300: + raise requests.HTTPError( + f"Callback returned status {response.status_code}", response=response + ) + return response + + @ft.singledispatch def load_description(filename: Path) -> str: if not filename.exists(): diff --git a/docs/wiki/cross-cutting/exceptions.md b/docs/wiki/cross-cutting/exceptions.md index 99da15713..57281a881 100644 --- a/docs/wiki/cross-cutting/exceptions.md +++ b/docs/wiki/cross-cutting/exceptions.md @@ -9,6 +9,9 @@ All paths relative to `backend/app/`. ## Provider/SDK error convention - `.claude/conventions/error-handling.md` — the standardized pattern for wrapping provider/SDK exceptions at service/crud call sites. Follow it for any new OpenAI/Gemini/Anthropic call site. +## Outbound retry policy +- `core/retry.py`: `is_transient(error)` (5xx or a connection that never reached upstream). Call sites use tenacity `@retry(stop=stop_after_attempt(4), retry=retry_if_exception(is_transient))` directly: `utils.send_callback`, `services/llm/jobs.py` and `services/llm/guardrails.py`. Batch providers keep their own policy. + ## Route-level rules - Routes return structured error codes + human-readable messages (see any module page's endpoint errors). - Per-row/batch work isolates failures: one item's provider error must not fail sibling items (pattern used across evaluations and batch processing). diff --git a/docs/wiki/modules/guardrails.md b/docs/wiki/modules/guardrails.md index a02672288..fd56bacfa 100644 --- a/docs/wiki/modules/guardrails.md +++ b/docs/wiki/modules/guardrails.md @@ -47,6 +47,8 @@ Returns `GuardrailsOutcome(safe_text, error, bypassed, rephrase_needed, raw)`. P `proxy_guardrails_request(method, path, ...)` is a separate, dumber path used only by the management CRUD routes: forwards verbatim, returns `(status_code, parsed_json_or_None)`. +**Retries** (tenacity `@retry`, 3 retries): validation POST and validator-config GETs retry on `is_transient` (5xx / connect failure); auth errors are never retried. The management proxy retries connect failures only, so a POST is never re-sent after it landed. After exhaustion each call falls into its §4 failure path. + ## 4. Failure semantics (the thing most changes get wrong) | Upstream condition | Behaviour | diff --git a/docs/wiki/modules/llm-call.md b/docs/wiki/modules/llm-call.md index 4d4249510..e95836167 100644 --- a/docs/wiki/modules/llm-call.md +++ b/docs/wiki/modules/llm-call.md @@ -36,6 +36,7 @@ All paths relative to `backend/app/`. ## Services / CRUD - `services/llm/` — `mappers.py` (Kaapi params → provider API), `providers/`, `chain/`, `guardrails.py`, `jobs.py` - `jobs.py::execute_llm_call` is the single shared invocation, reached from `/llm/call`, from `ChainBlock.execute`, and from fast evaluation runs. Two knobs shape what it records: `record_call` (default `True`) turns off the `LlmCall` row, the AI spans and the LLM metrics in one switch, for traffic that is not production. The switch lives on `_CallContext` (`ctx.span`, `ctx.metrics_started/finished`, and `_create_call_record` check it), so add a new span, metric or row write through `ctx` rather than calling the tracer or `record_llm_call_*` directly. Langfuse is turned off separately by passing `langfuse_credentials=None`. `BlockResult.guardrail_outcome` (`GuardrailOutcomeEnum.BLOCKED` / `.REPHRASED` / `None`, a `StrEnum` in `chain/types.py`) tells a caller a guardrail verdict apart from a provider failure, since both otherwise arrive as a bare string in `error` — see [guardrails.md](guardrails.md) §4, §6. `BlockResult.retryable` answers the other question that string cannot: whether re-running the identical call could succeed. It defaults to `False` and is set only where the provider or the proxy call actually failed, so a failure path added later fails fast instead of inheriting three attempts. +- **Retries** (tenacity `@retry`, 3 retries): `_post_to_proxy` retries 5xx/connect failures (never `ReadTimeout`). Those failures come back with `retryable=False` so fast-eval `retry_llm_call` doesn't stack more attempts. `utils.send_callback` retries any non-2xx or `requests.RequestException`, re-signing per attempt. Provider calls use the SDK's own retry: OpenAI/Anthropic `max_retries=3` on the client, Sarvam `request_options={"max_retries": 3}` per call. ElevenLabs STT/TTS use a tenacity `@retry` (4 attempts, 5xx/connect) in `eleven_ai.py` because the SDK `max_retries` is unusable; Gemini/Vertex retries live on the genai SDK branch. - `mappers.py` is the **only** param mapper in the codebase; the assessment fork was merged back into it. Callers: `crud/evaluations/{batch,fast,judge}.py`, `crud/assessment/batch.py`, `services/assessment/api/batch.py`, `services/assessment/prefilter/request_builder.py`. Structured output is keyed `output_schema` for every provider (OpenAI `text.format`, Anthropic `output_config.format`, Gemini `output_schema`), and Anthropic reads `effort` into `output_config.effort`. Every param that only assessment set (`top_p`, `max_output_tokens`, `thinking_level`, `thinking`, `output_schema`) defaults to `None` on `TextLLMParams` and is dropped by `ParamSerialization._dump_compact`, so a config that does not set it maps exactly as before. Anthropic never receives `temperature`/`top_p`: the Messages API returns 400 for a non-default sampling value on every Claude model Kaapi serves, so the mapper drops both, warning only when the value was not Anthropic's own default of 1.0. `normalize_llm_text` no longer lives here; it moved to `services/assessment/validators.py`, the only place that calls it. - `services/guardrails/` — validator execution - `crud/llm.py`, `crud/llm_chain.py`, `crud/config/` — persistence