feat(retry): Retry webhooks, LLM proxy, guardrails and provider SDK calls

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)
+    )

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 []

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:

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(

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. "

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(

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

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