fix: map models across provider fallbacks

Author: EvanProgramming

README.md

@@ -576,6 +576,15 @@ The local CLI is intentionally a high-permission agent, similar to Codex or Open
 }
 ```
 
+When automatic fallback is used, `auto`, an empty model, and the primary
+provider's configured simple/complex model names are mapped to the equivalent
+simple/complex model configured for each fallback provider. An explicit model
+name that is not recognised as one of the primary provider's names is passed
+through unchanged; use that only when the fallback endpoint supports the same
+name. The mapping is identical for streaming and non-streaming requests, and
+an all-provider failure retains every provider error with the primary failure
+as its cause.
+
 The file is auto-managed. Use `/provider` or `/api_key` in-chat to update it interactively.
 
 ---

providers.py

@@ -443,30 +443,73 @@ def __init__(self, primary_config: ProviderConfig) -> None:
     def config(self) -> ProviderConfig:
         return self._primary.config
 
+    def _model_for(self, provider: LLMProvider, requested_model: str | None) -> str:
+        """Choose a model that has the same simple/complex meaning for *provider*.
+
+        The model names in a provider configuration are provider-specific.  A
+        model selected from the primary provider must therefore not be sent to
+        a fallback merely because it is a non-empty string.  Names configured
+        for the primary (including its built-in defaults) are treated as
+        semantic simple/complex slots and mapped to the fallback's slots.
+        An unrecognised explicit name is forwarded as-is so users can use a
+        model name shared by several compatible endpoints, with the fallback
+        provider responsible for accepting or rejecting it.
+        """
+        primary_config = self._primary.config
+        target_config = provider.config
+        requested = (requested_model or "").strip()
+
+        if not requested or requested.lower() == "auto":
+            return target_config.model_simple
+
+        primary_simple = {
+            primary_config.model_simple,
+            PROVIDER_DEFAULT_MODELS.get(primary_config.provider, ("", ""))[0],
+        }
+        primary_complex = {
+            primary_config.model_complex,
+            PROVIDER_DEFAULT_MODELS.get(primary_config.provider, ("", ""))[1],
+        }
+        if requested in primary_complex and requested not in primary_simple:
+            return target_config.model_complex
+        if requested in primary_simple:
+            return target_config.model_simple
+        return requested
+
+    @staticmethod
+    def _raise_all_failed(attempts: list[tuple[str, Exception]]) -> None:
+        """Raise a useful error without discarding the primary failure."""
+        if not attempts:
+            raise RuntimeError("All providers failed")
+        if len(attempts) == 1:
+            raise attempts[0][1]
+        details = "; ".join(
+            f"{provider}: {type(error).__name__}: {error}"
+            for provider, error in attempts
+        )
+        error = RuntimeError(f"All providers failed ({details})")
+        raise error from attempts[0][1]
+
     def chat(self, messages: list[dict[str, str]], model: str | None = None) -> tuple[str, dict | None]:
         providers = [self._primary] + self._fallbacks
-        last_error = None
-        for i, prov in enumerate(providers):
+        attempts: list[tuple[str, Exception]] = []
+        for prov in providers:
             try:
-                return prov.chat(messages, model)
+                return prov.chat(messages, self._model_for(prov, model))
             except Exception as e:
-                last_error = e
-                if i < len(providers) - 1:
-                    continue  # try next
-        raise last_error or RuntimeError("All providers failed")
+                attempts.append((prov.name, e))
+        self._raise_all_failed(attempts)
 
     def chat_stream(self, messages: list[dict[str, str]], model: str | None = None) -> Iterator[str]:
         providers = [self._primary] + self._fallbacks
-        last_error = None
-        for i, prov in enumerate(providers):
+        attempts: list[tuple[str, Exception]] = []
+        for prov in providers:
             try:
-                yield from prov.chat_stream(messages, model)
+                yield from prov.chat_stream(messages, self._model_for(prov, model))
                 return
             except Exception as e:
-                last_error = e
-                if i < len(providers) - 1:
-                    continue
-        raise last_error or RuntimeError("All providers failed")
+                attempts.append((prov.name, e))
+        self._raise_all_failed(attempts)
 
     @property
     def name(self) -> str:

tests/test_providers.py

@@ -6,6 +6,8 @@
 from pathlib import Path
 
 from providers import (
+    FallbackProvider,
+    LLMProvider,
     ProviderConfig,
     _get_encryption_key,
     decrypt_api_key,
@@ -14,6 +16,29 @@
 )
 
 
+class _ProbeProvider(LLMProvider):
+    def __init__(self, provider: str, simple: str, complex_model: str,
+                 response: str = "ok", error: Exception | None = None) -> None:
+        super().__init__(ProviderConfig(provider=provider, model_simple=simple,
+                                         model_complex=complex_model))
+        self.models: list[str | None] = []
+        self.stream_models: list[str | None] = []
+        self.response = response
+        self.error = error
+
+    def chat(self, messages, model=None):
+        self.models.append(model)
+        if self.error:
+            raise self.error
+        return self.response, None
+
+    def chat_stream(self, messages, model=None):
+        self.stream_models.append(model)
+        if self.error:
+            raise self.error
+        yield self.response
+
+
 class ProviderConfigTests(unittest.TestCase):
     def setUp(self):
         self._home = os.environ.get("HOME")
@@ -48,5 +73,49 @@ def test_legacy_xor_ciphertext_remains_readable(self):
         self.assertEqual(decrypt_api_key(ciphertext), plaintext)
 
 
+class FallbackModelTests(unittest.TestCase):
+    def _fallback(self, primary, fallback):
+        wrapper = FallbackProvider.__new__(FallbackProvider)
+        wrapper._primary = primary
+        wrapper._fallbacks = [fallback]
+        return wrapper
+
+    def test_sync_maps_deepseek_complex_model_to_openai_default(self):
+        primary = _ProbeProvider("deepseek", "deepseek-chat", "deepseek-reasoner", error=RuntimeError("primary down"))
+        fallback = _ProbeProvider("openai", "gpt-4o", "gpt-4o", response="fallback")
+        result = self._fallback(primary, fallback).chat([], "deepseek-reasoner")
+        self.assertEqual(result[0], "fallback")
+        self.assertEqual(primary.models, ["deepseek-reasoner"])
+        self.assertEqual(fallback.models, ["gpt-4o"])
+
+    def test_stream_maps_anthropic_simple_model_to_deepseek_default(self):
+        primary = _ProbeProvider("anthropic", "claude-sonnet-4-20250514", "claude-sonnet-4-20250514",
+                                 error=RuntimeError("primary down"))
+        fallback = _ProbeProvider("deepseek", "deepseek-chat", "deepseek-reasoner", response="stream fallback")
+        result = list(self._fallback(primary, fallback).chat_stream([], "claude-sonnet-4-20250514"))
+        self.assertEqual(result, ["stream fallback"])
+        self.assertEqual(primary.stream_models, ["claude-sonnet-4-20250514"])
+        self.assertEqual(fallback.stream_models, ["deepseek-chat"])
+
+    def test_auto_and_unknown_explicit_models_have_documented_behavior(self):
+        primary = _ProbeProvider("deepseek", "deepseek-chat", "deepseek-reasoner", error=RuntimeError("primary down"))
+        fallback = _ProbeProvider("openai", "gpt-4o", "gpt-4o", response="fallback")
+        wrapper = self._fallback(primary, fallback)
+        wrapper.chat([], "auto")
+        wrapper.chat([], "shared-model")
+        self.assertEqual(fallback.models, ["gpt-4o", "shared-model"])
+
+    def test_all_provider_failures_keep_primary_error_as_cause_and_include_both(self):
+        primary_error = RuntimeError("primary outage")
+        fallback_error = RuntimeError("fallback outage")
+        primary = _ProbeProvider("deepseek", "deepseek-chat", "deepseek-reasoner", error=primary_error)
+        fallback = _ProbeProvider("openai", "gpt-4o", "gpt-4o", error=fallback_error)
+        with self.assertRaises(RuntimeError) as context:
+            self._fallback(primary, fallback).chat([], "auto")
+        self.assertIs(context.exception.__cause__, primary_error)
+        self.assertIn("primary outage", str(context.exception))
+        self.assertIn("fallback outage", str(context.exception))
+
+
 if __name__ == "__main__":
     unittest.main()