fix(reliability,observability,cli): retry adapter-wrapped transient errors, bound real timeouts, expose reliability state

Author: ctrl-gaurav

docs/observability/reliability.md

@@ -108,8 +108,15 @@ async def call_api_async() -> str:
 `is_transient_error` returns `True` for:
 - `ConnectionError`, `ConnectionResetError`, `OSError`
 - `TimeoutError`, `asyncio.TimeoutError`, `EffGenTimeoutError`
-- HTTP 429 (rate-limited) via `.status_code` or `.response.status_code`
-- HTTP 5xx (server errors)
+- Any error carrying a structured `.error_context` (every provider adapter
+  wraps a real SDK failure this way, so a live 429/5xx from any provider
+  classifies correctly even after wrapping — see `effgen.models.errors`)
+- A raw, never-wrapped SDK exception with HTTP 429 or 5xx via `.status_code`
+  or `.response.status_code`
+- `httpx` network/timeout errors
+
+A plain application exception with none of the above signals is *not*
+retried — retrying a bug in your own code will not fix it.
 
 ### Delay computation
 

effgen/cli/_main.py

@@ -3526,12 +3526,22 @@ def _handle_doctor_command(args) -> int:
     # System / CUDA / vLLM / pip-check report.
     system_report = _doctor_system_report(include_pip_check=live)
 
+    # Circuit-breaker/bulkhead state for any provider that has been routed
+    # through effgen.reliability middleware this process — surfaces an open
+    # circuit or a saturated bulkhead without the caller instrumenting their
+    # own code.
+    reliability_report = _doctor_reliability_report()
+
     # Exit nonzero if a live probe was requested and a keyed provider failed.
     # Computed once so every output format (JSON and human) agrees.
     exit_code = _doctor_exit_code(results, live)
 
     if getattr(args, 'output_json', False):
-        print(_json.dumps({"providers": results, "system": system_report}, indent=2))
+        print(_json.dumps({
+            "providers": results,
+            "system": system_report,
+            "reliability": reliability_report,
+        }, indent=2))
         return exit_code
 
     # Pretty-print
@@ -3585,6 +3595,32 @@ def _handle_doctor_command(args) -> int:
             sys_table.add_row(k, str(v))
         console.print(sys_table)
 
+        # Reliability section — only shown once a provider has actually been
+        # routed through circuit-breaker/bulkhead middleware this process.
+        if reliability_report:
+            console.print("\n[bold cyan]Reliability[/bold cyan]")
+            rel_table = Table(show_header=True)
+            rel_table.add_column("Provider", style="cyan", no_wrap=True)
+            rel_table.add_column("Circuit", style="white")
+            rel_table.add_column("Bulkhead", style="white")
+            for prov, rec in sorted(reliability_report.items()):
+                cb = rec.get("circuit_breaker")
+                bh = rec.get("bulkhead")
+                if cb is None:
+                    circuit_cell = "[dim]—[/dim]"
+                elif cb["state"] == "closed":
+                    circuit_cell = "[green]closed[/green]"
+                elif cb["state"] == "half_open":
+                    circuit_cell = "[yellow]half_open[/yellow]"
+                else:
+                    circuit_cell = "[red]open[/red]"
+                if bh is None:
+                    bulkhead_cell = "[dim]—[/dim]"
+                else:
+                    bulkhead_cell = f"active={bh['active']}/{bh['max_concurrency']}, queued={bh['queued']}/{bh['queue_size']}"
+                rel_table.add_row(prov, circuit_cell, bulkhead_cell)
+            console.print(rel_table)
+
         # Print hints for missing keys
         missing = [p for p, i in results.items() if not i.get("available")]
         if missing:
@@ -3609,6 +3645,17 @@ def _handle_doctor_command(args) -> int:
         print("\nSystem:")
         for k, v in system_report.items():
             print(f"  {k}: {v}")
+        if reliability_report:
+            print("\nReliability:")
+            for prov, rec in sorted(reliability_report.items()):
+                cb = rec.get("circuit_breaker")
+                bh = rec.get("bulkhead")
+                circuit_str = cb["state"] if cb else "—"
+                bulkhead_str = (
+                    f"active={bh['active']}/{bh['max_concurrency']}, queued={bh['queued']}/{bh['queue_size']}"
+                    if bh else "—"
+                )
+                print(f"  {prov:12s} circuit={circuit_str:10s} bulkhead={bulkhead_str}")
         missing = [p for p, i in results.items() if not i.get("available")]
         if missing:
             print("\nMissing keys — set in ~/.effgen/.env or export:")
@@ -3635,6 +3682,27 @@ def _doctor_exit_code(results: dict[str, dict], live: bool) -> int:
     return 0
 
 
+def _doctor_reliability_report() -> dict[str, dict]:
+    """Circuit-breaker/bulkhead state for providers routed through reliability
+    middleware this process, keyed by provider name (empty if none have).
+
+    A provider only appears once ``ProviderRegistry.get_circuit_breaker``/
+    ``get_bulkhead`` has been used for it — no calls yet made means no state
+    to report, which is the common case for a fresh CLI invocation.
+    """
+    try:
+        from effgen.models.registry import ProviderRegistry
+
+        stats = ProviderRegistry.reliability_stats()
+    except Exception:
+        return {}
+    return {
+        prov: rec
+        for prov, rec in stats.items()
+        if rec.get("circuit_breaker") is not None or rec.get("bulkhead") is not None
+    }
+
+
 def _doctor_system_report(*, include_pip_check: bool = False) -> dict[str, Any]:
     """Collect a CUDA / torch / vLLM / pip-check diagnostic snapshot."""
     report: dict[str, Any] = {}

effgen/models/_rate_limit.py

@@ -50,7 +50,19 @@ async def call():
 
 
 class RateLimitExceeded(Exception):
-    """Raised when the daily budget for a model is exhausted."""
+    """Raised when the daily budget for a model is exhausted.
+
+    Carries the same structured ``.error_context`` shape as
+    :mod:`effgen.models.errors`'s typed provider errors (category
+    ``"rate_limited"``) so ``is_transient_error()``/``classify_provider_error()``
+    classify it consistently with every other rate-limit signal.
+    """
+
+    def __init__(self, message: str) -> None:
+        super().__init__(message)
+        from effgen.models.errors import error_context_dict
+
+        self.error_context = error_context_dict("", "", "request", "rate_limited")
 
 
 @dataclass

effgen/models/routing/retry.py

@@ -6,12 +6,29 @@
   - ``RateLimitExceeded``  — provider rate-limited; back off and try again.
   - ``ProviderTransientError`` — 5xx from provider; may resolve on retry.
   - ``ModelTimeoutError``  — prediction timed out; worth retrying.
+  - ``BudgetExceededError`` — not retried against the *same* provider, but
+    treated as retriable here because the router's meaning of "retriable" is
+    "worth trying the next candidate" (e.g. a free-tier fallback), not "the
+    same call might succeed if repeated."
 
 Do NOT retry on (permanent, operator-action required):
   - ``ModelAuthError``     — bad/missing API key.
   - ``ModelRefusalError``  — model refused; retrying will produce the same result.
   - ``InvalidRequestError`` — malformed request; retrying will always fail.
 
+Relationship to ``effgen.models.errors.classify_provider_error`` /
+``effgen.reliability.retry.is_transient_error``: those answer "is retrying
+*this exact call* worth it", so an unrecognized exception defaults to
+retryable (a genuine transient blip should not become a hard failure) and
+``BudgetExceededError`` defaults to *not* retryable (repeating the same call
+against the same provider/budget will not succeed). ``RetryPolicy`` answers a
+related but distinct question — "should the router try the next candidate" —
+so it deliberately keeps its own explicit allow/deny lists rather than
+delegating: an unrecognized exception here defaults to *not* retriable
+(fail closed on an unknown failure rather than failing over blindly), and
+``BudgetExceededError`` is retriable here specifically because failing over
+to a different, cheaper/free-tier candidate is exactly the useful response.
+
 Usage::
 
     from effgen.models.routing.retry import RetryPolicy
@@ -104,7 +121,13 @@ def __init__(
     # ------------------------------------------------------------------
 
     def is_retriable(self, exc: Exception) -> bool:
-        """Return True if *exc* is safe to retry."""
+        """Return True if *exc* is worth trying the next candidate for.
+
+        Uses this module's own allow/deny lists rather than
+        ``classify_provider_error`` — see the module docstring for why the
+        two intentionally disagree on ``BudgetExceededError`` and on
+        unrecognized exceptions.
+        """
         if isinstance(exc, _NON_RETRIABLE):
             return False
         return isinstance(exc, _RETRIABLE)

effgen/models/transformers_engine.py

@@ -48,6 +48,21 @@
 logger = logging.getLogger(__name__)
 
 
+def _reraise_if_classified(exc: Exception) -> None:
+    """Re-raise *exc* unwrapped when it already carries retry classification.
+
+    A timeout raised by ``effgen.reliability.timeouts.with_timeout()`` around
+    a local generate call must propagate as-is instead of being flattened
+    into a generic ``RuntimeError`` by the callers' blanket exception
+    handlers below — flattening discards the type information
+    ``is_transient_error()`` relies on to retry it correctly.
+    """
+    from effgen.reliability.timeouts import TimeoutError as EffGenTimeoutError
+
+    if isinstance(exc, EffGenTimeoutError):
+        raise exc
+
+
 # Native tool-call delimiters that the downstream parser (core.tool_calling)
 # needs to see; these must survive the special-token strip on the tool path.
 _TOOL_CALL_DELIMITERS = frozenset({
@@ -716,6 +731,7 @@ def generate(
 
         except Exception as e:
             logger.error(f"Generation failed: {e}")
+            _reraise_if_classified(e)
             raise RuntimeError(f"Generation failed: {e}") from e
         finally:
             self._tokenizer_lock.release()
@@ -849,6 +865,7 @@ def _generate_with_error_capture() -> None:
 
         except Exception as e:
             logger.error(f"Streaming generation failed: {e}")
+            _reraise_if_classified(e)
             raise RuntimeError(f"Streaming generation failed: {e}") from e
 
     def generate_batch(
@@ -941,6 +958,7 @@ def generate_batch(
 
         except Exception as e:
             logger.error(f"Batch generation failed: {e}")
+            _reraise_if_classified(e)
             raise RuntimeError(f"Batch generation failed: {e}") from e
         finally:
             self._tokenizer_lock.release()