From 7a08103c1451b78a549eff64e51d907fea2ea93a Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 18:47:58 -0700 Subject: [PATCH] refactor(agent/auxiliary_client): split recovery ladder into rung helpers, unify call_llm twins --- agent/auxiliary_client.py | 1065 ++++++++++++++++++------------------- 1 file changed, 510 insertions(+), 555 deletions(-) diff --git a/agent/auxiliary_client.py b/agent/auxiliary_client.py index 09128562db..dee86229e0 100644 --- a/agent/auxiliary_client.py +++ b/agent/auxiliary_client.py @@ -8420,9 +8420,8 @@ def _resolve_call_client( main_runtime: Optional[Dict[str, Any]], async_mode: bool, ) -> _ResolvedAuxRoute: - """Resolve the client for one aux call: vision chain, or cached text client with - the explicit-provider fallback_chain / auto-chain rescue. Raises RuntimeError - with the user-facing setup hint when nothing is configured.""" + """Resolve the client for one aux call: vision chain, or cached text client with the + explicit-provider fallback_chain / auto-chain rescue; RuntimeError when nothing is configured.""" effective_provider = resolved_provider if task == "vision": effective_provider, client, final_model = resolve_vision_provider_client( @@ -8444,12 +8443,8 @@ def _resolve_call_client( async_mode=async_mode, main_runtime=main_runtime, ) - if client is None: - raise RuntimeError( - f"No LLM provider configured for task={task} provider={resolved_provider}. " - f"Run: hermes setup" - ) - resolved_provider = effective_provider or resolved_provider + if client is not None: + resolved_provider = effective_provider or resolved_provider else: client, final_model = _get_cached_client( resolved_provider, @@ -8465,27 +8460,26 @@ def _resolve_call_client( client, resolved_provider, ) if client is None: - # Explicit provider with no credentials: honor the task fallback_chain - # before raising (fallback entries may use OAuth / credential-pool auth). + # Explicit provider with no credentials: honor the task fallback_chain before + # raising (fallback entries may use OAuth / credential-pool auth). _explicit = (resolved_provider or "").strip().lower() if _explicit and _explicit not in {"auto", "openrouter", "custom"}: fb_client, fb_model, fb_label = _try_configured_fallback_for_unavailable_client( task, _explicit, ) - if fb_client is not None: - client, final_model = fb_client, fb_model - if async_mode: - client, final_model = _to_async_client( - fb_client, fb_model or "", is_vision=(task == "vision") - ) - resolved_provider = fb_label or resolved_provider - effective_provider = resolved_provider - else: + if fb_client is None: raise RuntimeError( f"Provider '{_explicit}' is set in config.yaml but no API key " f"was found. Set the {_explicit.upper()}_API_KEY environment " f"variable, or switch to a different provider with `hermes model`." ) + client, final_model = fb_client, fb_model + if async_mode: + client, final_model = _to_async_client( + fb_client, fb_model or "", is_vision=(task == "vision") + ) + resolved_provider = fb_label or resolved_provider + effective_provider = resolved_provider # Auto/custom with no credentials: walk the full auto chain (not just # OpenRouter). model=None so each provider uses its own default. if client is None and not resolved_base_url: @@ -8497,10 +8491,10 @@ def _resolve_call_client( effective_provider = _effective_provider_for_client( client, "auto", ) - if client is None: - raise RuntimeError( - f"No LLM provider configured for task={task} provider={resolved_provider}. " - f"Run: hermes setup") + if client is None: + raise RuntimeError( + f"No LLM provider configured for task={task} provider={resolved_provider}. " + f"Run: hermes setup") return _ResolvedAuxRoute(client, final_model, resolved_provider, effective_provider) @@ -8541,11 +8535,8 @@ def _prepare_aux_request( async_mode: bool, ) -> _PreparedAuxRequest: """Shared head of call_llm/async_call_llm: resolve route + client, publish it, build request kwargs. - - The sync wire additionally applies the certified compression fast lane and - per-request ``extra_headers``; ``base_info`` is the client's base_url (sync - falls back to the resolved base_url when the client exposes none). - """ + Sync-only: compression fast lane, per-request ``extra_headers``, and ``base_info`` falling + back to the resolved base_url when the client exposes none.""" resolved_provider, resolved_model, resolved_base_url, resolved_api_key, resolved_api_mode = _resolve_task_provider_model( task, provider, model, base_url, api_key) if api_mode: @@ -8593,8 +8584,8 @@ def _prepare_aux_request( task, request_provider or "auto", final_model or "default", f" at {base_info}" if base_info and "openrouter" not in base_info else "") - # Pass the client's actual base_url so endpoint-specific temperature overrides - # work on auto-detected routes (api.moonshot.ai vs api.kimi.com/coding). + # Client's actual base_url so endpoint-specific temperature overrides work on + # auto-detected routes (api.moonshot.ai vs api.kimi.com/coding). kwargs = _build_call_kwargs( request_provider, final_model, messages, temperature=temperature, max_tokens=max_tokens, @@ -8602,9 +8593,8 @@ def _prepare_aux_request( reasoning_config=reasoning_config, base_url=base_info or resolved_base_url, task=task) if fast_compression_cap is not None and max_tokens is None: - # Narrow exception to "no cap" on aux calls: the compression route is - # certified non-reasoning, so a bounded summary is intentional. Only fires - # when the caller passed no max_tokens (explicit caps pass through untouched). + # The compression route is certified non-reasoning, so a bounded summary is + # intentional; explicit caller caps pass through untouched. kwargs.update(auxiliary_max_tokens_param(fast_compression_cap, model=final_model)) if extra_headers: kwargs["extra_headers"] = dict(extra_headers) @@ -8622,11 +8612,8 @@ def _prepare_aux_request( class _LadderStep(NamedTuple): - """A provider request the ladder asks its driver to perform. - - kind: "call" (client, kwargs) | "retry_same_provider" (provider, model) | - "fallback" (fb_client, fb_model, fb_label). - """ + """A provider request the ladder asks its driver to perform. kind: "call" (client, kwargs) | + "retry_same_provider" (provider, model) | "fallback" (fb_client, fb_model, fb_label).""" kind: str args: tuple @@ -8666,6 +8653,292 @@ def _rung(step: "_LadderStep", accept: Callable[[Exception], bool]): return result, None +def _call_step(target_client: Any, request_kwargs: Dict[str, Any]) -> _LadderStep: + return _LadderStep("call", (target_client, request_kwargs)) + + +def _param_rung_accepts(exc: Exception) -> bool: + """After a parameter-strip retry: fall through to the max_tokens/payment/auth + chains with the stripped kwargs; re-raise anything those chains won't handle.""" + return ( + _is_payment_error(exc) + or _is_connection_error(exc) + or _is_auth_error(exc) + or "max_tokens" in str(exc) + or "unsupported_parameter" in str(exc) + ) + + +def _capacity_rung_accepts(exc: Exception) -> bool: + return _is_payment_error(exc) or _is_connection_error(exc) or _is_rate_limit_error(exc) + + +def _credential_rung_accepts(exc: Exception) -> bool: + return _is_auth_error(exc) or _is_payment_error(exc) or _is_rate_limit_error(exc) + + +# Immutable route context shared by the recovery rungs. +_LadderRoute = NamedTuple("_LadderRoute", [ + ("client", Any), ("task", Optional[str]), ("tag", str), ("async_mode", bool), ("base_info", str), + ("resolved_provider", str), ("resolved_model", Optional[str]), ("resolved_base_url", Optional[str]), + ("resolved_api_key", Optional[str]), ("resolved_api_mode", Optional[str]), + ("final_model", Optional[str]), ("main_runtime", Optional[Dict[str, Any]]), + ("route_info", Optional[Dict[str, str]]), +]) + + +def _ladder_parameter_rungs( + first_err: Exception, route: _LadderRoute, kwargs: Dict[str, Any], max_tokens: Optional[int], +): + """Rungs 1-3: retry without temperature / structured-output format / max_tokens. + Returns ``(response, None, kwargs)`` or ``(None, narrowed_err, stripped_kwargs)``.""" + client, task, tag = route.client, route.task, route.tag + if "temperature" in kwargs and _is_unsupported_temperature_error(first_err): + retry_kwargs = {k: v for k, v in kwargs.items() if k != "temperature"} + logger.info( + "Auxiliary %s%s: provider rejected temperature; retrying once without it", + task or "call", tag, + ) + resp, first_err = yield from _rung(_call_step(client, retry_kwargs), _param_rung_accepts) + if first_err is None: + return resp, None, retry_kwargs + kwargs = retry_kwargs + + if _is_structured_output_rejection(first_err): + retry_kwargs = _without_structured_output_format(kwargs) + if retry_kwargs is not None: + logger.info( + "Auxiliary %s%s: provider rejected the structured-output " + "format field; retrying once without it (schema " + "enforcement degrades to prompt compliance): %s", + task or "call", tag, first_err, + ) + resp, first_err = yield from _rung(_call_step(client, retry_kwargs), _param_rung_accepts) + if first_err is None: + return resp, None, retry_kwargs + kwargs = retry_kwargs + + err_str = str(first_err) + # ZAI vision models reject max_tokens with code 1210 and a message that + # never mentions "max_tokens", so detect it explicitly. + _is_zai_param_error = ( + "1210" in err_str + and "bigmodel" in str(getattr(client, "base_url", "")) + ) + if max_tokens is not None and ( + "max_tokens" in err_str + or "unsupported_parameter" in err_str + or _is_unsupported_parameter_error(first_err, "max_tokens") + or _is_zai_param_error + ): + kwargs.pop("max_tokens", None) + kwargs.pop("max_completion_tokens", None) + resp, first_err = yield from _rung(_call_step(client, kwargs), _capacity_rung_accepts) + if first_err is None: + return resp, None, kwargs + return None, first_err, kwargs + + +def _refreshed_nous_step( + route: _LadderRoute, kwargs: Dict[str, Any], message: str, +) -> Optional[_LadderStep]: + """Rebuild the Nous client after a credential event; None when nothing refreshed.""" + refreshed_client, refreshed_model = _refresh_nous_auxiliary_client( + cache_provider=route.resolved_provider or "nous", + model=route.final_model, + lookup_model=route.resolved_model, + lookup_task=route.task, + async_mode=route.async_mode, + base_url=route.resolved_base_url, + api_key=route.resolved_api_key, + api_mode=route.resolved_api_mode, + main_runtime=route.main_runtime, + is_vision=(route.task == "vision"), + ) + if refreshed_client is None: + return None + logger.info(message, route.task or "call", route.tag) + if refreshed_model and refreshed_model != kwargs.get("model"): + kwargs["model"] = refreshed_model + return _call_step(refreshed_client, kwargs) + + +def _ladder_nous_rungs( + first_err: Exception, route: _LadderRoute, kwargs: Dict[str, Any], client_is_nous: bool, +): + """Nous-only rungs: stale-model self-heal, paid-account refresh, 401 refresh. + Returns ``(response, None)`` or ``(None, first_err)`` to fall through.""" + client, task, tag = route.client, route.task, route.tag + # A long-lived process can pin a Portal model since dropped from the catalog + # (every call 404s); force a fresh Portal fetch and retry once. + if _is_model_not_found_error(first_err) and client_is_nous: + healed_model = _refresh_nous_recommended_model( + vision=(task == "vision"), stale_model=kwargs.get("model")) + if healed_model and healed_model != kwargs.get("model"): + logger.warning( + "Auxiliary %s%s: model %r no longer in Nous catalog; " + "retrying with refreshed recommendation %r", + task or "call", tag, kwargs.get("model"), healed_model, + ) + kwargs["model"] = healed_model + resp, first_err = yield from _rung(_call_step(client, kwargs), lambda exc: True) + if first_err is None: + return resp, None + + # Auth refresh parity with the main agent. + if ( + _is_payment_error(first_err) + and client_is_nous + and _nous_portal_account_has_fresh_paid_access() + ): + step = _refreshed_nous_step( + route, kwargs, + "Auxiliary %s%s: refreshed Nous runtime credentials after paid account check, retrying", + ) + if step is not None: + resp, first_err = yield from _rung( + step, lambda exc: _credential_rung_accepts(exc) or _is_connection_error(exc), + ) + if first_err is None: + return resp, None + + if _is_auth_error(first_err) and client_is_nous: + step = _refreshed_nous_step( + route, kwargs, + "Auxiliary %s%s: refreshed Nous runtime credentials after 401, retrying", + ) + if step is not None: + return (yield step), None + return None, first_err + + +def _ladder_credential_rungs( + first_err: Exception, route: _LadderRoute, kwargs: Dict[str, Any], client_is_nous: bool, +): + """OAuth credential refresh + same-provider retry, then credential-pool rotation. + Returns ``(response, None)`` or ``(None, first_err)`` to fall through.""" + client, task, tag, resolved_provider = route.client, route.task, route.tag, route.resolved_provider + auth_refresh_provider = _auth_refresh_provider_for_route( + resolved_provider, route.base_info) + if (_is_auth_error(first_err) + and auth_refresh_provider not in {"auto", "", None} + and not client_is_nous): + if _refresh_provider_credentials(auth_refresh_provider): + if auth_refresh_provider != _normalize_aux_provider(resolved_provider): + # The stale client is cached under the route label + # (e.g. "auto"), not the concrete backend we refreshed. + _evict_cached_clients(resolved_provider) + logger.info( + "Auxiliary %s%s: refreshed %s credentials after auth error, retrying", + task or "call", tag, auth_refresh_provider, + ) + return (yield _LadderStep( + "retry_same_provider", + (auth_refresh_provider, route.resolved_model or route.final_model), + )), None + + pool_provider = _recoverable_pool_provider(resolved_provider, client, main_runtime=route.main_runtime) + # Capture the exact key used so recovery finds the right pool entry even if + # another process rotated the pool meanwhile (current() would be None). + _client_api_key = str(getattr(client, "api_key", "") or "") + if pool_provider and _credential_rung_accepts(first_err): + recovery_err = first_err + # Skip the extra retry for clear payment/quota errors — the endpoint + # won't accept another request with the same exhausted key. + if _is_rate_limit_error(first_err) and not _is_payment_error(first_err): + resp, recovery_err = yield from _rung(_call_step(client, kwargs), _credential_rung_accepts) + if recovery_err is None: + return resp, None + if _recover_provider_pool(pool_provider, recovery_err, failed_api_key=_client_api_key): + logger.info( + "Auxiliary %s%s: recovered %s via credential-pool rotation after %s", + task or "call", tag, pool_provider, type(recovery_err).__name__, + ) + try: + return (yield _LadderStep( + "retry_same_provider", (resolved_provider, route.resolved_model), + )), None + except Exception as retry2_err: + # Rotated key also hit a wall: mark it now so concurrent processes + # skip it, then fall through to the provider fallback. + if (_is_payment_error(retry2_err) or _is_auth_error(retry2_err) + or _is_rate_limit_error(retry2_err)): + _recover_provider_pool(pool_provider, retry2_err) + first_err = retry2_err + else: + raise + return None, first_err + + +def _ladder_provider_fallback(first_err: Exception, route: _LadderRoute): + """Last rung: other providers (per-task chain; then auto: main fallback chain + + discovery chain, explicit: main-agent-model net). Returns the response or None. + + Capacity errors (payment/quota, connection, exhausted 429, model incompatible, + malformed response) bypass the explicit-provider gate — the provider cannot + serve this request regardless of user intent. Auth errors only fall back in auto mode. + """ + task, tag, resolved_provider = route.task, route.tag, route.resolved_provider + is_auto = resolved_provider in {"auto", "", None} + reason = _fallback_reason(first_err) + is_capacity_error = any( + predicate(first_err) for predicate, label in _FALLBACK_REASONS if label != "auth error" + ) + if reason is None or not (is_auto or is_capacity_error): + return None + if reason == "payment error": + # Mark the concrete backend (not the "auto" label) unhealthy so + # later aux calls skip it instead of paying another doomed RTT. + _mark_provider_unhealthy( + _recoverable_pool_provider(resolved_provider, route.client, main_runtime=route.main_runtime) + or resolved_provider + ) + logger.info("Auxiliary %s%s: %s on %s (%s), trying fallback", + task or "call", tag, reason, resolved_provider, first_err) + + # Skip only the failed model for model-specific failures; 401/402 are + # provider-wide, so keep skipping the whole provider. + _chain_failed_model = ( + None if reason in ("auth error", "payment error") else route.final_model + ) + fb_client, fb_model, fb_label = _try_configured_fallback_chain( + task, resolved_provider or "auto", reason=reason, + failed_model=_chain_failed_model) + if fb_client is None and is_auto: + fb_client, fb_model, fb_label = _try_main_fallback_chain( + task, resolved_provider or "auto", reason=reason) + if fb_client is None: + fb_client, fb_model, fb_label = _try_payment_fallback( + resolved_provider, task, reason=reason) + elif fb_client is None: + fb_client, fb_model, fb_label = _try_main_agent_model_fallback( + resolved_provider, task, reason=reason, + failed_model=_chain_failed_model) + + if fb_client is not None: + # Second pass: the candidate credential was stale and quarantined — walk + # the discovery chain once more (unhealthy entries are skipped). + for _pass in range(2): + _record_route_info( + route.route_info, _fallback_provider_from_label(fb_label), fb_model + ) + fb_resp = yield _LadderStep("fallback", (fb_client, fb_model, fb_label)) + if fb_resp is not None: + return fb_resp + if _pass == 0: + fb_client, fb_model, fb_label = _try_payment_fallback( + resolved_provider, task, reason="stale fallback credential") + if fb_client is None: + break + # All fallback layers exhausted — one user-visible warning, then re-raise. + logger.warning( + "Auxiliary %s%s: %s on %s and all fallbacks exhausted " + "(fallback_chain + main agent model). Raising original error.", + task or "call", tag, reason, resolved_provider, + ) + return None + + def _aux_recovery_ladder( first_err: Exception, *, @@ -8684,281 +8957,32 @@ def _aux_recovery_ladder( main_runtime: Optional[Dict[str, Any]], route_info: Optional[Dict[str, str]], ): - """Ordered recovery rungs after the primary request failed (generator). - - Rungs, in order: temperature strip → structured-output strip → max_tokens - strip → Nous stale-model self-heal → Nous paid/401 credential refresh → - OAuth credential refresh + same-provider retry → credential-pool rotation → - provider fallback (per-task chain, main fallback chain, discovery chain / - main-agent-model net). Each rung either returns a response, narrows - ``first_err`` and falls through, or re-raises. Returns ``_RERAISE_ORIGINAL`` - when every rung is exhausted (after evicting a connection-poisoned client). - """ + """Ordered recovery rungs after the primary request failed (generator): parameter + strips → Nous heal/refresh → credential refresh/pool rotation → provider fallback. + Each rung returns a response, narrows ``first_err`` and falls through, or re-raises. + Returns ``_RERAISE_ORIGINAL`` when exhausted (after evicting a connection-poisoned client).""" tag = " (async)" if async_mode else "" - - def _call(target_client: Any, request_kwargs: Dict[str, Any]) -> _LadderStep: - return _LadderStep("call", (target_client, request_kwargs)) - - - def _param_rung_accepts(exc: Exception) -> bool: - # Fall through to the max_tokens/payment/auth chains with the stripped - # kwargs; re-raise anything those chains won't handle. - return ( - _is_payment_error(exc) - or _is_connection_error(exc) - or _is_auth_error(exc) - or "max_tokens" in str(exc) - or "unsupported_parameter" in str(exc) - ) - - def _capacity_rung_accepts(exc: Exception) -> bool: - return _is_payment_error(exc) or _is_connection_error(exc) or _is_rate_limit_error(exc) - - def _credential_rung_accepts(exc: Exception) -> bool: - return _is_auth_error(exc) or _is_payment_error(exc) or _is_rate_limit_error(exc) - - if "temperature" in kwargs and _is_unsupported_temperature_error(first_err): - retry_kwargs = dict(kwargs) - retry_kwargs.pop("temperature", None) - logger.info( - "Auxiliary %s%s: provider rejected temperature; retrying once without it", - task or "call", tag, - ) - resp, first_err = yield from _rung(_call(client, retry_kwargs), _param_rung_accepts) - if first_err is None: - return resp - kwargs = retry_kwargs - - if _is_structured_output_rejection(first_err): - retry_kwargs = _without_structured_output_format(kwargs) - if retry_kwargs is not None: - logger.info( - "Auxiliary %s%s: provider rejected the structured-output " - "format field; retrying once without it (schema " - "enforcement degrades to prompt compliance): %s", - task or "call", tag, first_err, - ) - resp, first_err = yield from _rung(_call(client, retry_kwargs), _param_rung_accepts) - if first_err is None: - return resp - kwargs = retry_kwargs - - err_str = str(first_err) - # ZAI vision models reject max_tokens with code 1210 and a message that - # never mentions "max_tokens", so detect it explicitly. - _is_zai_param_error = ( - "1210" in err_str - and "bigmodel" in str(getattr(client, "base_url", "")) + route = _LadderRoute( + client, task, tag, async_mode, base_info, resolved_provider, resolved_model, + resolved_base_url, resolved_api_key, resolved_api_mode, final_model, + main_runtime, route_info, ) - if max_tokens is not None and ( - "max_tokens" in err_str - or "unsupported_parameter" in err_str - or _is_unsupported_parameter_error(first_err, "max_tokens") - or _is_zai_param_error - ): - kwargs.pop("max_tokens", None) - kwargs.pop("max_completion_tokens", None) - resp, first_err = yield from _rung(_call(client, kwargs), _capacity_rung_accepts) - if first_err is None: - return resp - - # ── Stale-model self-heal (Nous Portal recommendation drift) ─── - # A long-lived process can pin a Portal model since dropped from the catalog - # (every call 404s); force a fresh Portal fetch and retry once. Nous-only. - _heal_is_nous = ( - resolved_provider == "nous" - or base_url_host_matches(base_info, "inference-api.nousresearch.com") - ) - if _is_model_not_found_error(first_err) and _heal_is_nous: - healed_model = _refresh_nous_recommended_model( - vision=(task == "vision"), stale_model=kwargs.get("model")) - if healed_model and healed_model != kwargs.get("model"): - logger.warning( - "Auxiliary %s%s: model %r no longer in Nous catalog; " - "retrying with refreshed recommendation %r", - task or "call", tag, kwargs.get("model"), healed_model, - ) - kwargs["model"] = healed_model - resp, first_err = yield from _rung(_call(client, kwargs), lambda exc: True) - if first_err is None: - return resp - - # ── Nous auth refresh parity with main agent ────────────────── + resp, first_err, kwargs = yield from _ladder_parameter_rungs(first_err, route, kwargs, max_tokens) + if first_err is None: + return resp client_is_nous = ( resolved_provider == "nous" or base_url_host_matches(base_info, "inference-api.nousresearch.com") ) - if ( - _is_payment_error(first_err) - and client_is_nous - and _nous_portal_account_has_fresh_paid_access() - ): - refreshed_client, refreshed_model = _refresh_nous_auxiliary_client( - cache_provider=resolved_provider or "nous", - model=final_model, - lookup_model=resolved_model, - lookup_task=task, - async_mode=async_mode, - base_url=resolved_base_url, - api_key=resolved_api_key, - api_mode=resolved_api_mode, - main_runtime=main_runtime, - is_vision=(task == "vision"), - ) - if refreshed_client is not None: - logger.info( - "Auxiliary %s%s: refreshed Nous runtime credentials after paid account check, retrying", - task or "call", tag, - ) - if refreshed_model and refreshed_model != kwargs.get("model"): - kwargs["model"] = refreshed_model - resp, first_err = yield from _rung( - _call(refreshed_client, kwargs), - lambda exc: _credential_rung_accepts(exc) or _is_connection_error(exc), - ) - if first_err is None: - return resp - - if _is_auth_error(first_err) and client_is_nous: - refreshed_client, refreshed_model = _refresh_nous_auxiliary_client( - cache_provider=resolved_provider or "nous", - model=final_model, - lookup_model=resolved_model, - lookup_task=task, - async_mode=async_mode, - base_url=resolved_base_url, - api_key=resolved_api_key, - api_mode=resolved_api_mode, - main_runtime=main_runtime, - is_vision=(task == "vision"), - ) - if refreshed_client is not None: - logger.info("Auxiliary %s%s: refreshed Nous runtime credentials after 401, retrying", - task or "call", tag) - if refreshed_model and refreshed_model != kwargs.get("model"): - kwargs["model"] = refreshed_model - return (yield _call(refreshed_client, kwargs)) - - # ── Auth refresh retry ─────────────────────────────────────── - auth_refresh_provider = _auth_refresh_provider_for_route( - resolved_provider, base_info) - if (_is_auth_error(first_err) - and auth_refresh_provider not in {"auto", "", None} - and not client_is_nous): - if _refresh_provider_credentials(auth_refresh_provider): - if auth_refresh_provider != _normalize_aux_provider(resolved_provider): - # The stale client is cached under the route label - # (e.g. "auto"), not the concrete backend we refreshed. - _evict_cached_clients(resolved_provider) - logger.info( - "Auxiliary %s%s: refreshed %s credentials after auth error, retrying", - task or "call", tag, auth_refresh_provider, - ) - return (yield _LadderStep("retry_same_provider", (auth_refresh_provider, resolved_model or final_model))) - - # ── Same-provider credential-pool recovery ───────────────────── - pool_provider = _recoverable_pool_provider(resolved_provider, client, main_runtime=main_runtime) - # Capture the exact key used so recovery finds the right pool entry even if - # another process rotated the pool meanwhile (current() would be None). - _client_api_key = str(getattr(client, "api_key", "") or "") - if pool_provider and (_is_auth_error(first_err) or _is_payment_error(first_err) or _is_rate_limit_error(first_err)): - recovery_err = first_err - # Skip the extra retry for clear payment/quota errors — the endpoint - # won't accept another request with the same exhausted key. - if _is_rate_limit_error(first_err) and not _is_payment_error(first_err): - resp, recovery_err = yield from _rung(_call(client, kwargs), _credential_rung_accepts) - if recovery_err is None: - return resp - if _recover_provider_pool(pool_provider, recovery_err, failed_api_key=_client_api_key): - logger.info( - "Auxiliary %s%s: recovered %s via credential-pool rotation after %s", - task or "call", tag, pool_provider, type(recovery_err).__name__, - ) - try: - return (yield _LadderStep("retry_same_provider", (resolved_provider, resolved_model))) - except Exception as retry2_err: - # Rotated key also hit a wall: mark it now so concurrent processes - # skip it, then fall through to the payment fallback below. - if (_is_payment_error(retry2_err) or _is_auth_error(retry2_err) - or _is_rate_limit_error(retry2_err)): - _recover_provider_pool(pool_provider, retry2_err) - first_err = retry2_err - else: - raise - - # ── Payment / connection / rate-limit / auth fallback ───────── - # Try alternative providers when the resolved one returns 402/credit - # exhaustion, is unreachable, is rate-limited (429), or 401s past the - # refresh paths. Auth is NOT a capacity error: it only bypasses the - # explicit-provider gate in auto mode. - # Capacity errors (payment/quota, connection, exhausted 429, model incompatible - # with route, malformed response) bypass the explicit-provider gate: the - # provider cannot serve this request regardless of user intent. Auth errors - # are NOT capacity errors: they only fall back in auto mode. - is_auto = resolved_provider in {"auto", "", None} - reason = _fallback_reason(first_err) - is_capacity_error = any( - predicate(first_err) for predicate, label in _FALLBACK_REASONS if label != "auth error" - ) - if reason is not None and (is_auto or is_capacity_error): - if reason == "payment error": - # Mark the concrete backend (not the "auto" label) unhealthy so - # later aux calls skip it instead of paying another doomed RTT. - _mark_provider_unhealthy( - _recoverable_pool_provider(resolved_provider, client, main_runtime=main_runtime) or resolved_provider - ) - logger.info("Auxiliary %s%s: %s on %s (%s), trying fallback", - task or "call", tag, reason, resolved_provider, first_err) - - # Skip only the failed model for model-specific failures; 401/402 are - # provider-wide, so keep skipping the whole provider. - _chain_failed_model = ( - None if reason in ("auth error", "payment error") else final_model - ) - # Fallback order: per-task fallback_chain; then for auto: main - # fallback_providers, then built-in discovery chain; for explicit - # providers: main agent model safety net. - fb_client, fb_model, fb_label = (None, None, "") - if is_auto: - fb_client, fb_model, fb_label = _try_configured_fallback_chain( - task, resolved_provider or "auto", reason=reason, - failed_model=_chain_failed_model) - if fb_client is None: - fb_client, fb_model, fb_label = _try_main_fallback_chain( - task, resolved_provider or "auto", reason=reason) - if fb_client is None: - fb_client, fb_model, fb_label = _try_payment_fallback( - resolved_provider, task, reason=reason) - else: - fb_client, fb_model, fb_label = _try_configured_fallback_chain( - task, resolved_provider or "auto", reason=reason, - failed_model=_chain_failed_model) - if fb_client is None: - fb_client, fb_model, fb_label = _try_main_agent_model_fallback( - resolved_provider, task, reason=reason, - failed_model=_chain_failed_model) - - if fb_client is not None: - # Second pass: the candidate credential was stale and quarantined — walk - # the discovery chain once more (unhealthy entries are skipped). - for _pass in range(2): - _record_route_info( - route_info, _fallback_provider_from_label(fb_label), fb_model - ) - fb_resp = yield _LadderStep("fallback", (fb_client, fb_model, fb_label)) - if fb_resp is not None: - return fb_resp - if _pass == 0: - fb_client, fb_model, fb_label = _try_payment_fallback( - resolved_provider, task, reason="stale fallback credential") - if fb_client is None: - break - # All fallback layers exhausted — one user-visible warning, then re-raise. - logger.warning( - "Auxiliary %s%s: %s on %s and all fallbacks exhausted " - "(fallback_chain + main agent model). Raising original error.", - task or "call", tag, reason, resolved_provider, - ) + resp, first_err = yield from _ladder_nous_rungs(first_err, route, kwargs, client_is_nous) + if first_err is None: + return resp + resp, first_err = yield from _ladder_credential_rungs(first_err, route, kwargs, client_is_nous) + if first_err is None: + return resp + resp = yield from _ladder_provider_fallback(first_err, route) + if resp is not None: + return resp # Connection/timeout errors poison the cached client (closed transport, # half-read stream); evict so the next aux call rebuilds a fresh one. if _is_connection_error(first_err): @@ -9000,6 +9024,17 @@ async def _drive_ladder_async(ladder, perform: Callable[[_LadderStep], Any]) -> return stop.value +def _elapsed_ms(started_at: float, now: Optional[float] = None) -> int: + """Whole milliseconds since ``started_at`` (clamped at 0).""" + return max(0, int(((time.monotonic() if now is None else now) - started_at) * 1000)) + + +def _stamp_latency_once(latency_info: Optional[Dict[str, int]], key: str, started_at: float) -> None: + """Record ``key`` in ``latency_info`` the first time it fires.""" + if latency_info is not None and key not in latency_info: + latency_info[key] = _elapsed_ms(started_at) + + @_relay_auxiliary_call def call_llm( task: str = None, @@ -9030,23 +9065,9 @@ def call_llm( semaphore.acquire() request_started_at = time.monotonic() if latency_info is not None: - latency_info["queue_wait_ms"] = max( - 0, int((request_started_at - queue_started_at) * 1000) - ) + latency_info["queue_wait_ms"] = _elapsed_ms(queue_started_at, request_started_at) prior_progress_hook = getattr(_aux_progress, "hook", None) - def _timed_response() -> None: - if latency_info is not None and "time_to_first_progress_ms" not in latency_info: - latency_info["time_to_first_progress_ms"] = max( - 0, int((time.monotonic() - request_started_at) * 1000) - ) - - def _timed_dispatch() -> None: - if latency_info is not None and "provider_dispatch_ms" not in latency_info: - latency_info["provider_dispatch_ms"] = max( - 0, int((time.monotonic() - request_started_at) * 1000) - ) - try: with ( aux_progress_hook( @@ -9054,28 +9075,17 @@ def call_llm( if callable(prior_progress_hook) else ((lambda: None) if latency_info is not None else None) ), - _aux_timing_hook(_aux_dispatch, _timed_dispatch), - _aux_timing_hook(_aux_provider_response, _timed_response), + _aux_timing_hook(_aux_dispatch, functools.partial( + _stamp_latency_once, latency_info, "provider_dispatch_ms", request_started_at)), + _aux_timing_hook(_aux_provider_response, functools.partial( + _stamp_latency_once, latency_info, "time_to_first_progress_ms", request_started_at)), ): response = _call_llm_impl( - task=task, - provider=provider, - model=model, - base_url=base_url, - api_key=api_key, - main_runtime=main_runtime, - messages=messages, - temperature=temperature, - max_tokens=max_tokens, - tools=tools, - timeout=timeout, - extra_body=extra_body, - reasoning_config=reasoning_config, - extra_headers=extra_headers, - api_mode=api_mode, - stream=stream, - stream_options=stream_options, - route_info=route_info, + task=task, provider=provider, model=model, base_url=base_url, api_key=api_key, + main_runtime=main_runtime, messages=messages, temperature=temperature, + max_tokens=max_tokens, tools=tools, timeout=timeout, extra_body=extra_body, + reasoning_config=reasoning_config, extra_headers=extra_headers, api_mode=api_mode, + stream=stream, stream_options=stream_options, route_info=route_info, ) if stream and semaphore is not None: stream_semaphore = semaphore @@ -9084,9 +9094,7 @@ def call_llm( return response finally: if latency_info is not None: - latency_info["summary_generation_ms"] = max( - 0, int((time.monotonic() - request_started_at) * 1000) - ) + latency_info["summary_generation_ms"] = _elapsed_ms(request_started_at) if semaphore is not None: semaphore.release() @@ -9106,6 +9114,98 @@ def _release_sync_semaphore_after_stream( semaphore.release() +def _plan_aux_call( + task: Optional[str], + *, + async_mode: bool, + provider: Optional[str], + model: Optional[str], + base_url: Optional[str], + api_key: Optional[str], + main_runtime: Optional[Dict[str, Any]], + messages: list, + temperature: Optional[float], + max_tokens: Optional[int], + tools: Optional[list], + timeout: Optional[float], + extra_body: Optional[dict], + reasoning_config: Optional[dict], + extra_headers: Optional[Dict[str, str]], + api_mode: Optional[str], + route_info: Optional[Dict[str, str]], +) -> Tuple[_PreparedAuxRequest, Dict[str, Any], Dict[str, Any]]: + """Shared head of both call impls: prepare the request and bundle the kwargs the + recovery drivers pass to ``_retry_same_provider_*`` / ``_call_fallback_candidate_*``. + One immutable runtime snapshot for keying/resolution/retries/fallbacks, so a + concurrent /model switch can't mix key and client from different runtimes.""" + main_runtime = _normalize_main_runtime(main_runtime) + req = _prepare_aux_request( + task, provider=provider, model=model, base_url=base_url, api_key=api_key, + main_runtime=main_runtime, messages=messages, temperature=temperature, + max_tokens=max_tokens, tools=tools, timeout=timeout, extra_body=extra_body, + reasoning_config=reasoning_config, extra_headers=extra_headers, + api_mode=api_mode, route_info=route_info, async_mode=async_mode, + ) + candidate_kwargs = dict( + task=task, messages=messages, temperature=temperature, max_tokens=max_tokens, + tools=tools, effective_timeout=req.effective_timeout, + effective_extra_body=req.effective_extra_body, reasoning_config=reasoning_config, + ) + retry_kwargs = dict( + candidate_kwargs, resolved_base_url=req.resolved_base_url, + resolved_api_key=req.resolved_api_key, resolved_api_mode=req.resolved_api_mode, + main_runtime=main_runtime, final_model=req.final_model, extra_headers=extra_headers, + ) + return req, retry_kwargs, candidate_kwargs + + +def _should_retry_same_provider(task: Optional[str], exc: Exception, tag: str) -> bool: + """True when ``exc`` is a transient transport blip worth a same-provider retry. + + Critical-path tasks skip the retry on a full-budget timeout (see + _should_skip_same_provider_retry) and go straight to fallback. + """ + if not _is_transient_transport_error(exc): + return False + if _should_skip_same_provider_retry(task, exc): + logger.info( + "Auxiliary %s%s: timeout on the critical path; " + "skipping same-provider retry and falling back: %s", + task, tag, exc, + ) + return False + return True + + +def _ladder_step_call( + step: _LadderStep, req: _PreparedAuxRequest, retry_kwargs: Dict[str, Any], candidate_kwargs: Dict[str, Any], +) -> Tuple[str, tuple, Dict[str, Any]]: + """Resolve a ladder step into ``(kind, args, kwargs)`` for the sync/async performer.""" + if step.kind == "call": + return "call", step.args, dict(provider=req.resolved_provider, api_mode=req.resolved_api_mode) + if step.kind == "retry_same_provider": + retry_provider, retry_model = step.args + return "retry", (), dict(retry_kwargs, resolved_provider=retry_provider, resolved_model=retry_model) + return "fallback", step.args, candidate_kwargs + + +def _start_recovery_ladder( + first_err: Exception, req: _PreparedAuxRequest, retry_kwargs: Dict[str, Any], *, + task: Optional[str], async_mode: bool, route_info: Optional[Dict[str, str]], +): + """Build the recovery-ladder generator for a failed primary request.""" + return _aux_recovery_ladder( + first_err, + client=req.client, kwargs=req.kwargs, task=task, async_mode=async_mode, + base_info=req.base_info, + resolved_provider=req.resolved_provider, resolved_model=req.resolved_model, + resolved_base_url=req.resolved_base_url, resolved_api_key=req.resolved_api_key, + resolved_api_mode=req.resolved_api_mode, final_model=req.final_model, + max_tokens=retry_kwargs["max_tokens"], main_runtime=retry_kwargs["main_runtime"], + route_info=route_info, + ) + + def _call_llm_impl( task: str = None, *, @@ -9128,87 +9228,53 @@ def _call_llm_impl( route_info: Optional[Dict[str, str]] = None, ) -> Any: """Centralized synchronous LLM call: resolve provider/model, auth, kwargs, fallbacks. - - task: aux task name whose provider:model is read from config (ignored if provider set). - api_mode overrides task config; timeout=None reads auxiliary.{task}.timeout; - extra_headers override client defaults (e.g. Copilot ``x-initiator``). - stream=True returns the raw SDK stream iterator (caller consumes/falls back) instead - of a validated response. Raises RuntimeError if no provider is configured. - """ - # One immutable runtime snapshot for keying/resolution/retries/fallbacks, so a - # concurrent /model switch can't mix key and client from different runtimes. - main_runtime = _normalize_main_runtime(main_runtime) - req = _prepare_aux_request( - task, provider=provider, model=model, base_url=base_url, api_key=api_key, - main_runtime=main_runtime, messages=messages, temperature=temperature, - max_tokens=max_tokens, tools=tools, timeout=timeout, extra_body=extra_body, - reasoning_config=reasoning_config, extra_headers=extra_headers, - api_mode=api_mode, route_info=route_info, async_mode=False, + task: aux task whose provider:model comes from config (ignored if provider set); api_mode + overrides task config; timeout=None reads auxiliary.{task}.timeout; extra_headers override + client defaults. stream=True returns the raw SDK stream (caller consumes/falls back) + instead of a validated response. RuntimeError if no provider is configured.""" + req, retry_kwargs, candidate_kwargs = _plan_aux_call( + task, async_mode=False, provider=provider, model=model, base_url=base_url, + api_key=api_key, main_runtime=main_runtime, messages=messages, + temperature=temperature, max_tokens=max_tokens, tools=tools, timeout=timeout, + extra_body=extra_body, reasoning_config=reasoning_config, + extra_headers=extra_headers, api_mode=api_mode, route_info=route_info, ) - client, final_model, kwargs = req.client, req.final_model, req.kwargs - resolved_provider, request_provider = req.resolved_provider, req.request_provider - resolved_model, resolved_base_url = req.resolved_model, req.resolved_base_url - resolved_api_key, resolved_api_mode = req.resolved_api_key, req.resolved_api_mode - effective_timeout, effective_extra_body = req.effective_timeout, req.effective_extra_body - _base_info = req.base_info + client, kwargs, request_provider = req.client, req.kwargs, req.request_provider - # Streaming path (MoA aggregator): return the raw SDK stream, deliberately - # skipping validation and the fallback chain below — those assume a complete - # response. The caller owns reassembly, stale-stream detection and fallback. + # Streaming path (MoA aggregator): return the raw SDK stream, skipping validation and + # the fallback chain (they assume a complete response); the caller owns reassembly/fallback. if stream: kwargs["stream"] = True if stream_options: kwargs["stream_options"] = stream_options if task == "moa_aggregator" and isinstance(client, CodexAuxiliaryClient): - # Responses-shim clients consume the stream internally and return a - # completed object; Relay's managed stream would iterate that object - # itself. Return directly — the MoA facade wraps it as a one-chunk stream. + # Responses-shim clients consume the stream internally and return a completed + # object Relay's managed stream would iterate; the MoA facade wraps it as one chunk. return client.chat.completions.create(**kwargs) - return _relay_sync_stream( - client, - kwargs, - provider=request_provider, - api_mode=resolved_api_mode, - ) + return _relay_sync_stream(client, kwargs, provider=request_provider, api_mode=req.resolved_api_mode) def _primary(**validate_kw: Any) -> Any: return _validate_llm_response( _relay_sync_completion( - client, - kwargs, - provider=request_provider, - api_mode=resolved_api_mode, + client, kwargs, provider=request_provider, api_mode=req.resolved_api_mode, create=lambda request: _create_with_progress( - client, - request, - task, + client, request, task, force_stream=_provider_requires_stream( - request_provider, _base_info or resolved_base_url, + request_provider, req.base_info or req.resolved_base_url, ), ), ), - task, - **validate_kw, + task, **validate_kw, ) try: - # Bounded same-provider retry (exponential backoff, count from - # auxiliary.transient_retries) for transient transport blips before the - # except-chain escalates to fallback — a dropped connection shouldn't + # Bounded same-provider retry (exponential backoff, auxiliary.transient_retries) for + # transient blips before escalating to fallback — a dropped connection shouldn't # abandon a healthy provider (matters for pinned MoA advisors). try: - return _primary(provider=request_provider, base_url=_base_info) + return _primary(provider=request_provider, base_url=req.base_info) except Exception as transient_err: - if not _is_transient_transport_error(transient_err): - raise - # Critical-path tasks skip the same-provider retry on a - # full-budget timeout; see _should_skip_same_provider_retry. - if _should_skip_same_provider_retry(task, transient_err): - logger.info( - "Auxiliary %s: timeout on the critical path; " - "skipping same-provider retry and falling back: %s", - task, transient_err, - ) + if not _should_retry_same_provider(task, transient_err, ""): raise _max_transient_retries = _transient_retry_count() _last_transient = transient_err @@ -9230,52 +9296,15 @@ def _call_llm_impl( raise _last_transient except Exception as first_err: def _perform(step: _LadderStep) -> Any: - if step.kind == "call": - target_client, request_kwargs = step.args - return _validate_llm_response( - _relay_sync_completion( - target_client, request_kwargs, - provider=resolved_provider, api_mode=resolved_api_mode, - ), task) - if step.kind == "retry_same_provider": - retry_provider, retry_model = step.args - return _retry_same_provider_sync( - task=task, - resolved_provider=retry_provider, - resolved_model=retry_model, - resolved_base_url=resolved_base_url, - resolved_api_key=resolved_api_key, - resolved_api_mode=resolved_api_mode, - main_runtime=main_runtime, - final_model=final_model, - messages=messages, - temperature=temperature, - max_tokens=max_tokens, - tools=tools, - effective_timeout=effective_timeout, - effective_extra_body=effective_extra_body, - reasoning_config=reasoning_config, - extra_headers=extra_headers, - ) - fb_client, fb_model, fb_label = step.args - return _call_fallback_candidate_sync( - fb_client, fb_model, fb_label, - task=task, messages=messages, - temperature=temperature, max_tokens=max_tokens, - tools=tools, effective_timeout=effective_timeout, - effective_extra_body=effective_extra_body, - reasoning_config=reasoning_config) + kind, args, kw = _ladder_step_call(step, req, retry_kwargs, candidate_kwargs) + if kind == "call": + return _validate_llm_response(_relay_sync_completion(*args, **kw), task) + if kind == "retry": + return _retry_same_provider_sync(**kw) + return _call_fallback_candidate_sync(*args, **kw) result = _drive_ladder( - _aux_recovery_ladder( - first_err, - client=client, kwargs=kwargs, task=task, async_mode=False, - base_info=_base_info, - resolved_provider=resolved_provider, resolved_model=resolved_model, - resolved_base_url=resolved_base_url, resolved_api_key=resolved_api_key, - resolved_api_mode=resolved_api_mode, final_model=final_model, - max_tokens=max_tokens, main_runtime=main_runtime, route_info=route_info, - ), + _start_recovery_ladder(first_err, req, retry_kwargs, task=task, async_mode=False, route_info=route_info), _perform, ) if result is _RERAISE_ORIGINAL: @@ -9295,15 +9324,13 @@ def _coerce_llm_message(response): if "choices" not in response: return response choices = response.get("choices") or [] + else: + choices = getattr(response, "choices", None) if not choices: - return None - first = choices[0] - return first.get("message") if isinstance(first, dict) else getattr(first, "message", None) - choices = getattr(response, "choices", None) + return response if not choices: - return response - first = choices[0] - return first.get("message") if isinstance(first, dict) else getattr(first, "message", None) + return None + return _message_field(choices[0], "message") def _message_field(msg, name): @@ -9321,8 +9348,6 @@ def extract_content_or_reasoning(response, *, max_reasoning_chars: int | None = fallback so unbounded chain-of-thought can't become the compaction summary. Returns ``""`` if nothing found. """ - import re - msg = _coerce_llm_message(response) if msg is None: return "" @@ -9402,20 +9427,10 @@ async def async_call_llm( await semaphore.acquire() try: return await _async_call_llm_impl( - task=task, - provider=provider, - model=model, - base_url=base_url, - api_key=api_key, - main_runtime=main_runtime, - messages=messages, - temperature=temperature, - max_tokens=max_tokens, - tools=tools, - timeout=timeout, - extra_body=extra_body, - reasoning_config=reasoning_config, - route_info=route_info, + task=task, provider=provider, model=model, base_url=base_url, api_key=api_key, + main_runtime=main_runtime, messages=messages, temperature=temperature, + max_tokens=max_tokens, tools=tools, timeout=timeout, extra_body=extra_body, + reasoning_config=reasoning_config, route_info=route_info, ) finally: if semaphore is not None: @@ -9440,34 +9455,22 @@ async def _async_call_llm_impl( route_info: Optional[Dict[str, str]] = None, ) -> Any: """Centralized asynchronous LLM call; see call_llm() for full documentation.""" - # Keep every async phase on one runtime identity across awaits (concurrent /model switch). - main_runtime = _normalize_main_runtime(main_runtime) - extra_headers = None # async entry point has no per-request header override - req = _prepare_aux_request( - task, provider=provider, model=model, base_url=base_url, api_key=api_key, - main_runtime=main_runtime, messages=messages, temperature=temperature, - max_tokens=max_tokens, tools=tools, timeout=timeout, extra_body=extra_body, - reasoning_config=reasoning_config, extra_headers=None, - api_mode=None, route_info=route_info, async_mode=True, + # No per-request header / api_mode override on the async entry point. + req, retry_kwargs, candidate_kwargs = _plan_aux_call( + task, async_mode=True, provider=provider, model=model, base_url=base_url, + api_key=api_key, main_runtime=main_runtime, messages=messages, + temperature=temperature, max_tokens=max_tokens, tools=tools, timeout=timeout, + extra_body=extra_body, reasoning_config=reasoning_config, + extra_headers=None, api_mode=None, route_info=route_info, ) - client, final_model, kwargs = req.client, req.final_model, req.kwargs - resolved_provider, request_provider = req.resolved_provider, req.request_provider - resolved_model, resolved_base_url = req.resolved_model, req.resolved_base_url - resolved_api_key, resolved_api_mode = req.resolved_api_key, req.resolved_api_mode - effective_timeout, effective_extra_body = req.effective_timeout, req.effective_extra_body - _client_base = req.base_info + client, kwargs, request_provider = req.client, req.kwargs, req.request_provider try: - # Retry ONCE on the same provider for a transient blip before escalating - # to fallback — see call_llm() for the rationale. + # Retry ONCE on the same provider for a transient blip before fallback (see call_llm()). _force_stream_async = ( - _provider_requires_stream( - request_provider, _client_base or resolved_base_url, - ) + _provider_requires_stream(request_provider, req.base_info or req.resolved_base_url) and not isinstance(client, ( - AsyncCodexAuxiliaryClient, - AsyncAnthropicAuxiliaryClient, - AsyncBedrockAuxiliaryClient, + AsyncCodexAuxiliaryClient, AsyncAnthropicAuxiliaryClient, AsyncBedrockAuxiliaryClient, )) ) @@ -9479,29 +9482,17 @@ async def _async_call_llm_impl( async def _primary(**validate_kw: Any) -> Any: return _validate_llm_response( await _relay_async_completion( - client, - kwargs, - provider=request_provider, - api_mode=resolved_api_mode, + client, kwargs, provider=request_provider, api_mode=req.resolved_api_mode, create=_acreate, ), - task, - **validate_kw, + task, **validate_kw, ) try: - return await _primary(provider=request_provider, base_url=_client_base) + return await _primary(provider=request_provider, base_url=req.base_info) except Exception as transient_err: - if not _is_transient_transport_error(transient_err): - raise - # Same rule as call_llm(); the async Codex adapter wraps the sync - # stream via to_thread, so the same TimeoutError reaches here. - if _should_skip_same_provider_retry(task, transient_err): - logger.info( - "Auxiliary %s (async): timeout on the critical " - "path; skipping same-provider retry and falling back: %s", - task, transient_err, - ) + # The async Codex adapter wraps the sync stream via to_thread: same TimeoutError here. + if not _should_retry_same_provider(task, transient_err, " (async)"): raise logger.info( "Auxiliary %s (async): transient transport error; retrying " @@ -9511,55 +9502,19 @@ async def _async_call_llm_impl( return await _primary() except Exception as first_err: async def _perform(step: _LadderStep) -> Any: - if step.kind == "call": - target_client, request_kwargs = step.args - return _validate_llm_response( - await _relay_async_completion( - target_client, request_kwargs, - provider=resolved_provider, api_mode=resolved_api_mode, - ), task) - if step.kind == "retry_same_provider": - retry_provider, retry_model = step.args - return await _retry_same_provider_async( - task=task, - resolved_provider=retry_provider, - resolved_model=retry_model, - resolved_base_url=resolved_base_url, - resolved_api_key=resolved_api_key, - resolved_api_mode=resolved_api_mode, - main_runtime=main_runtime, - final_model=final_model, - messages=messages, - temperature=temperature, - max_tokens=max_tokens, - tools=tools, - effective_timeout=effective_timeout, - effective_extra_body=effective_extra_body, - reasoning_config=reasoning_config, - extra_headers=extra_headers, - ) - fb_client, fb_model, fb_label = step.args + kind, args, kw = _ladder_step_call(step, req, retry_kwargs, candidate_kwargs) + if kind == "call": + return _validate_llm_response(await _relay_async_completion(*args, **kw), task) + if kind == "retry": + return await _retry_same_provider_async(**kw) + fb_client, fb_model, fb_label = args fb_client, _ = _to_async_client( fb_client, fb_model or "", is_vision=(task == "vision") ) - return await _call_fallback_candidate_async( - fb_client, fb_model, fb_label, - task=task, messages=messages, - temperature=temperature, max_tokens=max_tokens, - tools=tools, effective_timeout=effective_timeout, - effective_extra_body=effective_extra_body, - reasoning_config=reasoning_config) + return await _call_fallback_candidate_async(fb_client, fb_model, fb_label, **kw) result = await _drive_ladder_async( - _aux_recovery_ladder( - first_err, - client=client, kwargs=kwargs, task=task, async_mode=True, - base_info=_client_base, - resolved_provider=resolved_provider, resolved_model=resolved_model, - resolved_base_url=resolved_base_url, resolved_api_key=resolved_api_key, - resolved_api_mode=resolved_api_mode, final_model=final_model, - max_tokens=max_tokens, main_runtime=main_runtime, route_info=route_info, - ), + _start_recovery_ladder(first_err, req, retry_kwargs, task=task, async_mode=True, route_info=route_info), _perform, ) if result is _RERAISE_ORIGINAL: