Spaces:
Sleeping
Sleeping
| """Configured auxiliary fallback-chain selection helpers (HermesFace overlay). | |
| Extracted so failure-injection tests can assert provider advancement without | |
| importing the full upstream ``agent.auxiliary_client`` module. | |
| """ | |
| from __future__ import annotations | |
| from typing import Callable, Iterable, Iterator, List, Tuple | |
| def should_skip_configured_fallback_entry( | |
| fb_provider: str, | |
| failed_provider: str, | |
| is_unhealthy: Callable[[str], bool], | |
| ) -> Tuple[bool, str]: | |
| """Return whether to skip a fallback entry and a short reason label.""" | |
| provider = str(fb_provider or "").strip() | |
| if not provider: | |
| return True, "empty" | |
| norm = provider.lower() | |
| skip = str(failed_provider or "").strip().lower() | |
| if norm == skip: | |
| return True, "failed" | |
| if is_unhealthy(norm): | |
| return True, "unhealthy" | |
| return False, "" | |
| def iter_configured_fallback_candidates( | |
| chain: Iterable[object], | |
| failed_provider: str, | |
| is_unhealthy: Callable[[str], bool], | |
| ) -> Iterator[Tuple[int, dict, str]]: | |
| """Yield viable (index, entry, label) tuples in chain order.""" | |
| if not isinstance(chain, list): | |
| return | |
| for index, entry in enumerate(chain): | |
| if not isinstance(entry, dict): | |
| continue | |
| fb_provider = str(entry.get("provider", "")).strip() | |
| skip, _reason = should_skip_configured_fallback_entry( | |
| fb_provider, failed_provider, is_unhealthy | |
| ) | |
| if skip: | |
| continue | |
| label = f"fallback_chain[{index}]({fb_provider})" | |
| yield index, entry, label | |
| def resolve_failed_provider_label( | |
| resolved_provider: str, | |
| *, | |
| recoverable_provider: str | None = None, | |
| ) -> str: | |
| """Prefer the concrete backend label over ``auto`` for chain skipping.""" | |
| if recoverable_provider: | |
| return recoverable_provider | |
| label = str(resolved_provider or "").strip() | |
| return label or "auto" | |
| def advance_after_simulated_failures( | |
| chain: List[dict], | |
| failed_providers: List[str], | |
| is_unhealthy: Callable[[str], bool], | |
| ) -> Tuple[int, dict, str] | None: | |
| """Return the next candidate after sequential provider failures in one call.""" | |
| failed_set = {str(p).strip().lower() for p in failed_providers if str(p).strip()} | |
| combined_unhealthy = lambda label: is_unhealthy(label) or label in failed_set | |
| failed_label = failed_providers[-1] if failed_providers else "auto" | |
| for index, entry, label in iter_configured_fallback_candidates( | |
| chain, failed_label, combined_unhealthy | |
| ): | |
| return index, entry, label | |
| return None | |