fix(transport): normalize lifecycle incident failures

Author: Reflex

src/runloop_api_client/_base_client.py

@@ -94,6 +94,7 @@
     APIResponseValidationError,
 )
 from ._utils._json import openapi_dumps
+from .lib.error_contract import is_safe_transport_retry
 
 log: logging.Logger = logging.getLogger(__name__)
 
@@ -665,7 +666,10 @@ def _enforce_trailing_slash(self, url: URL) -> URL:
     def _make_status_error_from_response(
         self,
         response: httpx.Response,
+        *,
+        attempts: int = 1,
     ) -> APIStatusError:
+        body: object | None
         if response.is_closed and not response.is_stream_consumed:
             # We can't read the response body as it has been closed
             # before it was read. This can happen if an event hook
@@ -677,12 +681,19 @@ def _make_status_error_from_response(
             body = err_text
 
             try:
-                body = json.loads(err_text)
-                err_msg = f"Error code: {response.status_code} - {body}"
+                body = cast(object, json.loads(err_text))
+                body_mapping = cast(Mapping[str, object], body) if isinstance(body, dict) else None
+                body_message = body_mapping.get("message") if body_mapping is not None else None
+                if isinstance(body_message, str):
+                    err_msg = body_message
+                else:
+                    err_msg = f"Error code: {response.status_code} - {body}"
             except Exception:
                 err_msg = err_text or f"Error code: {response.status_code}"
 
-        return self._make_status_error(err_msg, body=body, response=response)
+        error = self._make_status_error(err_msg, body=cast(object, body), response=response)
+        error.attempts = attempts
+        return error
 
     def _make_status_error(
         self,
@@ -1045,6 +1056,26 @@ def _should_retry(self, response: httpx.Response) -> bool:
             log.debug("Not retrying as header `x-should-retry` is set to `false`")
             return False
 
+        # These failures happen after a request or response may have been
+        # partially transferred. Retrying them implicitly can duplicate an
+        # execute or replay a multipart stream. Servers may explicitly opt in
+        # with X-Should-Retry when an idempotency record makes that safe.
+        try:
+            raw_payload = response.json()
+        except Exception:
+            raw_payload = None
+        payload = cast(Mapping[str, object], raw_payload) if isinstance(raw_payload, dict) else None
+        code = response.headers.get("x-runloop-error-code")
+        if code is None and isinstance(payload, dict) and isinstance(payload.get("error"), str):
+            code = payload["error"]
+        if code in {
+            "upload_request_body_idle_timeout",
+            "tunnel_backend_idle_timeout",
+            "tunnel_backend_connection_reset",
+        }:
+            log.debug("Not retrying ambiguous transfer failure %s", code)
+            return False
+
         # Retry on request timeouts.
         if response.status_code == 408:
             log.debug("Retrying due to status code %i", response.status_code)
@@ -1366,7 +1397,7 @@ def request(
             except httpx.TimeoutException as err:
                 log.debug("Encountered httpx.TimeoutException", exc_info=True)
 
-                if remaining_retries > 0:
+                if remaining_retries > 0 and is_safe_transport_retry(err):
                     self._sleep_for_retry(
                         retries_taken=retries_taken,
                         max_retries=max_retries,
@@ -1377,11 +1408,11 @@ def request(
                     continue
 
                 log.debug("Raising timeout error")
-                raise APITimeoutError(request=request) from err
+                raise APITimeoutError(request=request, cause=err, attempts=retries_taken + 1) from err
             except Exception as err:
                 log.debug("Encountered Exception", exc_info=True)
 
-                if remaining_retries > 0:
+                if remaining_retries > 0 and is_safe_transport_retry(err):
                     self._sleep_for_retry(
                         retries_taken=retries_taken,
                         max_retries=max_retries,
@@ -1392,7 +1423,7 @@ def request(
                     continue
 
                 log.debug("Raising connection error")
-                raise APIConnectionError(request=request) from err
+                raise APIConnectionError(request=request, cause=err, attempts=retries_taken + 1) from err
 
             log.debug(
                 'HTTP Response: %s %s "%i %s" %s',
@@ -1424,7 +1455,7 @@ def request(
                     err.response.read()
 
                 log.debug("Re-raising status error")
-                raise self._make_status_error_from_response(err.response) from None
+                raise self._make_status_error_from_response(err.response, attempts=retries_taken + 1) from None
 
             break
 
@@ -2076,7 +2107,7 @@ async def request(
             except httpx.TimeoutException as err:
                 log.debug("Encountered httpx.TimeoutException", exc_info=True)
 
-                if remaining_retries > 0:
+                if remaining_retries > 0 and is_safe_transport_retry(err):
                     await self._sleep_for_retry(
                         retries_taken=retries_taken,
                         max_retries=max_retries,
@@ -2087,11 +2118,11 @@ async def request(
                     continue
 
                 log.debug("Raising timeout error")
-                raise APITimeoutError(request=request) from err
+                raise APITimeoutError(request=request, cause=err, attempts=retries_taken + 1) from err
             except Exception as err:
                 log.debug("Encountered Exception", exc_info=True)
 
-                if remaining_retries > 0:
+                if remaining_retries > 0 and is_safe_transport_retry(err):
                     await self._sleep_for_retry(
                         retries_taken=retries_taken,
                         max_retries=max_retries,
@@ -2102,7 +2133,7 @@ async def request(
                     continue
 
                 log.debug("Raising connection error")
-                raise APIConnectionError(request=request) from err
+                raise APIConnectionError(request=request, cause=err, attempts=retries_taken + 1) from err
 
             log.debug(
                 'HTTP Response: %s %s "%i %s" %s',
@@ -2134,7 +2165,7 @@ async def request(
                     await err.response.aread()
 
                 log.debug("Re-raising status error")
-                raise self._make_status_error_from_response(err.response) from None
+                raise self._make_status_error_from_response(err.response, attempts=retries_taken + 1) from None
 
             break
 

src/runloop_api_client/_exceptions.py

@@ -6,6 +6,8 @@
 
 import httpx
 
+from .lib.error_contract import status_error_details, transport_error_details
+
 __all__ = [
     "BadRequestError",
     "AuthenticationError",
@@ -37,11 +39,39 @@ class APIError(RunloopError):
     If there was no response associated with this error then it will be `None`.
     """
 
-    def __init__(self, message: str, request: httpx.Request, *, body: object | None) -> None:  # noqa: ARG002
+    code: str
+    phase: str
+    retryable: bool
+    request_id: str | None
+    retry_after: float | None
+    attempts: int
+    cause: BaseException | None
+
+    def __init__(
+        self,
+        message: str,
+        request: httpx.Request,
+        *,
+        body: object | None,
+        code: str = "runloop_error",
+        phase: str = "unknown",
+        retryable: bool = False,
+        request_id: str | None = None,
+        retry_after: float | None = None,
+        attempts: int = 1,
+        cause: BaseException | None = None,
+    ) -> None:
         super().__init__(message)
         self.request = request
         self.message = message
         self.body = body
+        self.code = code
+        self.phase = phase
+        self.retryable = retryable
+        self.request_id = request_id
+        self.retry_after = retry_after
+        self.attempts = attempts
+        self.cause = cause
 
 
 class APIResponseValidationError(APIError):
@@ -60,20 +90,52 @@ class APIStatusError(APIError):
     response: httpx.Response
     status_code: int
 
-    def __init__(self, message: str, *, response: httpx.Response, body: object | None) -> None:
-        super().__init__(message, response.request, body=body)
+    def __init__(self, message: str, *, response: httpx.Response, body: object | None, attempts: int = 1) -> None:
+        details = status_error_details(response, body)
+        super().__init__(
+            message,
+            response.request,
+            body=body,
+            code=details.code,
+            phase=details.phase,
+            retryable=details.retryable,
+            request_id=details.request_id,
+            retry_after=details.retry_after,
+            attempts=attempts,
+        )
         self.response = response
         self.status_code = response.status_code
 
 
 class APIConnectionError(APIError):
-    def __init__(self, *, message: str = "Connection error.", request: httpx.Request) -> None:
-        super().__init__(message, request, body=None)
+    def __init__(
+        self,
+        *,
+        message: str = "Connection error.",
+        request: httpx.Request,
+        cause: BaseException | None = None,
+        attempts: int = 1,
+    ) -> None:
+        details = transport_error_details(cause) if cause is not None else transport_error_details(Exception())
+        super().__init__(
+            message,
+            request,
+            body=None,
+            code=details.code,
+            phase=details.phase,
+            retryable=details.retryable,
+            attempts=attempts,
+            cause=cause,
+        )
 
 
 class APITimeoutError(APIConnectionError):
-    def __init__(self, request: httpx.Request) -> None:
-        super().__init__(message="Request timed out.", request=request)
+    def __init__(self, request: httpx.Request, *, cause: BaseException | None = None, attempts: int = 1) -> None:
+        super().__init__(message="Request timed out.", request=request, cause=cause, attempts=attempts)
+        if cause is None:
+            self.code = "connection_timeout"
+            self.phase = "connect"
+            self.retryable = True
 
 
 class BadRequestError(APIStatusError):

src/runloop_api_client/lib/error_contract.py

@@ -0,0 +1,97 @@
+"""Stable normalization for Runloop API and HTTPX transport failures.
+
+This module is handwritten and intentionally lives under ``lib`` so generated
+client updates only need a small integration point.
+"""
+
+from __future__ import annotations
+
+from typing import Mapping, cast
+from dataclasses import dataclass
+
+import httpx
+
+
+@dataclass(frozen=True)
+class ErrorDetails:
+    code: str
+    phase: str
+    retryable: bool
+    request_id: str | None = None
+    retry_after: float | None = None
+
+
+def _number(value: object) -> float | None:
+    try:
+        parsed = float(value)  # type: ignore[arg-type]
+    except (TypeError, ValueError):
+        return None
+    return parsed if parsed >= 0 else None
+
+
+def parse_retry_after(headers: httpx.Headers, body: object = None) -> float | None:
+    """Parse Retry-After while accepting the SDK's millisecond extension."""
+    milliseconds = _number(headers.get("retry-after-ms"))
+    if milliseconds is not None:
+        return milliseconds / 1000
+    seconds = _number(headers.get("retry-after"))
+    if seconds is not None:
+        return seconds
+    if isinstance(body, Mapping):
+        payload = cast(Mapping[str, object], body)
+        details = payload.get("details")
+        if isinstance(details, Mapping):
+            return _number(cast(Mapping[str, object], details).get("retry_after"))
+    return None
+
+
+def status_error_details(response: httpx.Response, body: object) -> ErrorDetails:
+    payload: Mapping[str, object] = cast(Mapping[str, object], body) if isinstance(body, Mapping) else {}
+    header_code = response.headers.get("x-runloop-error-code")
+    body_code = payload.get("error")
+    code = header_code or (body_code if isinstance(body_code, str) else None) or f"http_{response.status_code}"
+    body_phase = payload.get("phase")
+    phase = body_phase if isinstance(body_phase, str) else "api"
+    retryable_value = payload.get("retryable")
+    retryable = retryable_value if isinstance(retryable_value, bool) else response.status_code in {408, 409, 429}
+    if response.status_code >= 500 and not isinstance(retryable_value, bool):
+        retryable = True
+    request_id: str | None = response.headers.get("x-runloop-request-id")
+    body_request_id = payload.get("request_id")
+    if request_id is None and isinstance(body_request_id, str):
+        request_id = body_request_id
+    return ErrorDetails(
+        code=code,
+        phase=phase,
+        retryable=retryable,
+        request_id=request_id,
+        retry_after=parse_retry_after(response.headers, cast(object, body)),
+    )
+
+
+def transport_error_details(error: BaseException) -> ErrorDetails:
+    if isinstance(error, httpx.ConnectTimeout):
+        return ErrorDetails("connection_timeout", "connect", True)
+    if isinstance(error, httpx.WriteTimeout):
+        return ErrorDetails("request_write_timeout", "request_write", False)
+    if isinstance(error, httpx.WriteError):
+        return ErrorDetails("request_write_failed", "request_write", False)
+    if isinstance(error, httpx.ReadTimeout):
+        return ErrorDetails("response_read_timeout", "response_read", False)
+    if isinstance(error, httpx.RemoteProtocolError):
+        if "idle_timeout" in str(error).lower():
+            return ErrorDetails("http2_idle_timeout", "response_read", False)
+        return ErrorDetails("http2_protocol_error", "transport", False)
+    if isinstance(error, httpx.TimeoutException):
+        return ErrorDetails("connection_timeout", "connect", True)
+    return ErrorDetails("connection_failed", "connect", isinstance(error, httpx.ConnectError))
+
+
+def is_safe_transport_retry(error: BaseException) -> bool:
+    """Only retry failures that prove the request body was not partially sent."""
+    # Preserve the generated client's handling of non-HTTPX exceptions (for
+    # example a pre-send auth hook failure). HTTPX errors carry enough phase
+    # information for the stricter partial-write audit below.
+    if not isinstance(error, httpx.HTTPError):
+        return True
+    return isinstance(error, (httpx.ConnectTimeout, httpx.ConnectError))