Clarify model routing precedence for Codex, OpenAI, and custom Kimi endpoints

docs/model-routing-precedence.md

@@ -0,0 +1,28 @@
+# Model Routing Precedence
+
+Pantheon now resolves chat model choices in one explicit order so that
+`codex`, `openai`, and configured Kimi coding endpoints do not silently
+override one another.
+
+## Final precedence
+
+```mermaid
+flowchart TD
+    A["User model choice"] --> B{"Explicit provider/model?"}
+    B -->|Yes| C["Use exact route<br/>Examples: codex/gpt-5.4-mini<br/>openai/gpt-5.4<br/>custom_anthropic/K2.5"]
+    B -->|No| D{"Known alias?"}
+    D -->|Yes| E["Normalize alias to explicit route<br/>codex oauth -> codex/gpt-5.4-mini<br/>openai chatgpt -> openai/gpt-5.4<br/>kimi-for-coding -> custom_anthropic/<configured model>"]
+    D -->|No| F{"Quality/capability tag?"}
+    F -->|Yes| G["Resolve tag via ModelSelector<br/>high / normal / low / vision / tools"]
+    F -->|No| H["Treat as explicit model name and validate provider"]
+    G --> I["Automatic provider selection"]
+    I --> J["For non-custom tag requests, skip custom endpoints and fall back to standard providers"]
+```
+
+## Why this matters
+
+- Explicit provider routes always win.
+- Friendly aliases are converted once into explicit routes and then persisted.
+- Tags still support automatic provider selection.
+- Custom endpoints are only chosen automatically for explicit custom requests,
+  not for unrelated `high` or `normal` tag resolutions.

pantheon/agent.py

@@ -144,9 +144,14 @@ def _resolve_model_tag(tag: str) -> list[str]:
 def _normalize_model_spec(
     model: str | list[str] | None,
 ) -> str | list[str] | None:
-    """Treat empty string model specs as unspecified."""
-    if isinstance(model, str) and not model.strip():
-        return None
+    """Treat empty string model specs as unspecified and canonicalize aliases."""
+    if isinstance(model, str):
+        from .utils.model_selector import normalize_model_choice
+
+        normalized = normalize_model_choice(model)
+        if not normalized:
+            return None
+        return normalized
     return model
 
 

pantheon/chatroom/room.py

@@ -2100,7 +2100,7 @@ async def set_agent_model(
         """
         try:
             from pantheon.agent import _is_model_tag, _resolve_model_tag, _parse_thinking_suffix
-            from pantheon.utils.model_selector import get_model_selector
+            from pantheon.utils.model_selector import normalize_model_choice
 
             # 1. Get team and find target agent
             team = await self.get_team_for_chat(chat_id)
@@ -2116,6 +2116,8 @@ async def set_agent_model(
 
             # 2. Parse +think suffix (e.g. "high+think:medium" → thinking="medium")
             clean_model, thinking = _parse_thinking_suffix(model)
+            clean_model = normalize_model_choice(clean_model)
+            persisted_model = f"{clean_model}+think:{thinking}" if thinking else clean_model
 
             # 3. Validate provider if requested
             if validate:
@@ -2164,7 +2166,7 @@ async def set_agent_model(
                         agent_name_lower = agent_name.lower()
                         for agent_cfg in original_team.agents:
                             if (agent_cfg.name or "").lower() == agent_name_lower or (agent_cfg.id or "").lower() == agent_name_lower:
-                                agent_cfg.model = model  # Store original input (tag or model name)
+                                agent_cfg.model = persisted_model
                                 break
 
                         # Write back to template file
@@ -2183,9 +2185,7 @@ async def set_agent_model(
             # Update the agent's model in template (case-insensitive match)
             for agent_config in team_template.get("agents", []):
                 if (agent_config.get("name") or "").lower() == agent_name_lower or (agent_config.get("id") or "").lower() == agent_name_lower:
-                    agent_config["model"] = (
-                        model  # Store original input (tag or model name)
-                    )
+                    agent_config["model"] = persisted_model
                     break
 
             memory.set_metadata("team_template", team_template)
@@ -2197,7 +2197,7 @@ async def set_agent_model(
             return {
                 "success": True,
                 "agent": agent_name,
-                "model": model,
+                "model": persisted_model,
                 "resolved_models": resolved_models,
             }
 
@@ -2277,12 +2277,13 @@ def _validate_model_provider(self, model: str) -> tuple[bool, str]:
             (is_valid, error_message)
         """
         from pantheon.agent import _is_model_tag
-        from pantheon.utils.model_selector import get_model_selector
+        from pantheon.utils.model_selector import get_model_selector, normalize_model_choice
 
         # Tags are always valid (they resolve based on available providers)
         if _is_model_tag(model):
             return True, ""
 
+        model = normalize_model_choice(model)
         selector = get_model_selector()
         available = selector._get_available_providers()
 

pantheon/utils/model_selector.py

@@ -68,6 +68,18 @@ class CustomEndpointConfig:
     ),
 }
 
+# Friendly aliases used by UI/users. They are intentionally narrow so that
+# switching among Codex OAuth, OpenAI API, and the configured Kimi coding
+# endpoint follows a single explicit route instead of falling back implicitly.
+EXPLICIT_MODEL_ALIASES: dict[str, str] = {
+    "codex": "codex/gpt-5.4-mini",
+    "codexoauth": "codex/gpt-5.4-mini",
+    "codexchatgpt": "codex/gpt-5.4-mini",
+    "chatgpt": "openai/gpt-5.4",
+    "openaichatgpt": "openai/gpt-5.4",
+    "openaichat": "openai/gpt-5.4",
+}
+
 # Sentinel object for negative cache (better than empty string)
 _NOT_FOUND = object()
 
@@ -77,6 +89,75 @@ class CustomEndpointConfig:
 _ollama_cache_time: float = 0
 
 
+def _canonicalize_alias_key(value: str) -> str:
+    """Normalize a user-facing alias key for comparison."""
+    return "".join(ch for ch in value.strip().lower() if ch.isalnum())
+
+
+def _get_settings_api_value(settings: "Settings | None", key: str) -> str:
+    """Read an env-like setting value from Settings when available."""
+    if settings is None or not hasattr(settings, "get_api_key"):
+        return ""
+    try:
+        value = settings.get_api_key(key)
+    except Exception:
+        return ""
+    return str(value or "").strip()
+
+
+def resolve_custom_endpoint_model(
+    provider_key: str,
+    settings: "Settings | None" = None,
+) -> str | None:
+    """Return the configured explicit model route for a custom endpoint."""
+    import os
+
+    config = CUSTOM_ENDPOINT_ENVS.get(provider_key)
+    if config is None:
+        return None
+
+    model = os.environ.get(config.model_env, "").strip()
+    if not model:
+        model = _get_settings_api_value(settings, config.model_env)
+    if not model:
+        return None
+
+    return f"{provider_key}/{model}"
+
+
+def normalize_model_choice(
+    model: str,
+    settings: "Settings | None" = None,
+) -> str:
+    """Canonicalize friendly aliases into explicit provider/model routes.
+
+    Precedence after normalization becomes:
+    1. Explicit provider/model routes
+    2. Known provider aliases mapped to explicit routes
+    3. Quality/capability tags
+    4. Automatic provider selection
+    """
+    normalized = model.strip()
+    if not normalized:
+        return normalized
+
+    alias_key = _canonicalize_alias_key(normalized)
+    explicit = EXPLICIT_MODEL_ALIASES.get(alias_key)
+    if explicit:
+        return explicit
+
+    if alias_key in {"kimiforcoding", "kimiforcode", "kimicoding"}:
+        explicit = resolve_custom_endpoint_model("custom_anthropic", settings=settings)
+        if explicit:
+            return explicit
+        logger.warning(
+            "Alias '{}' requested but CUSTOM_ANTHROPIC_MODEL is not configured; keeping original value",
+            model,
+        )
+
+    return normalized
+
+
 def _detect_ollama(base_url: str = "http://localhost:11434") -> bool:
     """Check if Ollama is running locally."""
     try:
@@ -920,4 +1001,7 @@ def get_default_model() -> list[str]:
     "FALLBACK_TAG",
     "CUSTOM_ENDPOINT_ENVS",
     "CustomEndpointConfig",
+    "EXPLICIT_MODEL_ALIASES",
+    "normalize_model_choice",
+    "resolve_custom_endpoint_model",
 ]

tests/test_agent.py

@@ -129,6 +129,17 @@ def test_blank_model_is_treated_as_implicit_default():
     assert agent.models
 
 
+def test_agent_normalizes_explicit_provider_aliases(monkeypatch):
+    monkeypatch.setattr(
+        "pantheon.utils.model_selector.normalize_model_choice",
+        lambda model, settings=None: "codex/gpt-5.4-mini" if model == "codex oauth" else model,
+    )
+
+    agent = Agent(name="alias", instructions="x", model="codex oauth")
+
+    assert agent.models == ["codex/gpt-5.4-mini"]
+
+
 async def test_call_agent_prefers_current_provider_for_quality_tag(monkeypatch):
     captured = {}
 

tests/test_model_selector.py

@@ -11,13 +11,16 @@
     CAPABILITY_MAP,
     DEFAULT_PROVIDER_MODELS,
     DEFAULT_PROVIDER_PRIORITY,
+    EXPLICIT_MODEL_ALIASES,
     ULTIMATE_FALLBACK,
     FALLBACK_TAG,
     QUALITY_TAGS,
     ModelSelector,
     get_default_model,
     get_model_selector,
+    normalize_model_choice,
     reset_model_selector,
+    resolve_custom_endpoint_model,
 )
 
 
@@ -27,6 +30,7 @@ def mock_settings():
     settings = MagicMock()
     # Mock get() to return the default value when key is not found
     settings.get.side_effect = lambda key, default=None: default
+    settings.get_api_key.side_effect = lambda key, default=None: default
     return settings
 
 
@@ -182,6 +186,25 @@ def test_resolve_unknown_capability_returns_quality_models(self, mock_settings):
         # Should just return normal models since unknown_capability is not in CAPABILITY_MAP
         assert len(models) > 0
 
+    def test_explicit_aliases_normalize_to_routed_models(self):
+        assert EXPLICIT_MODEL_ALIASES["codexoauth"] == "codex/gpt-5.4-mini"
+        assert normalize_model_choice("codex oauth") == "codex/gpt-5.4-mini"
+        assert normalize_model_choice("openai chatgpt") == "openai/gpt-5.4"
+
+    def test_kimi_alias_normalizes_to_configured_custom_endpoint(self, mock_settings):
+        mock_settings.get_api_key.side_effect = (
+            lambda key, default=None: "K2.5" if key == "CUSTOM_ANTHROPIC_MODEL" else default
+        )
+
+        assert (
+            resolve_custom_endpoint_model("custom_anthropic", settings=mock_settings)
+            == "custom_anthropic/K2.5"
+        )
+        assert (
+            normalize_model_choice("kimi-for-coding", settings=mock_settings)
+            == "custom_anthropic/K2.5"
+        )
+
 
 class TestCapabilityFiltering:
     """Test capability-based model filtering."""