feat(kimi): discover K3 on coding endpoint

agent/model_metadata.py

@@ -539,6 +539,29 @@ def _is_known_provider_base_url(base_url: str) -> bool:
     return _infer_provider_from_url(base_url) is not None
 
 
+def _endpoint_scoped_context_length(model: str, base_url: str) -> Optional[int]:
+    """Return metadata confirmed only for one provider endpoint."""
+    normalized = _normalize_base_url(base_url)
+    try:
+        parsed = urlparse(normalized)
+        port = parsed.port
+    except ValueError:
+        return None
+    if (
+        parsed.scheme.lower() == "https"
+        and (parsed.hostname or "").lower() == "api.kimi.com"
+        and port in (None, 443)
+        and parsed.username is None
+        and parsed.password is None
+        and parsed.path.rstrip("/") in {"/coding", "/coding/v1"}
+        and not parsed.query
+        and not parsed.fragment
+        and model.strip().lower() == "k3"
+    ):
+        return 1_048_576
+    return None
+
+
 def _skip_persistent_context_cache(base_url: str, provider: str) -> bool:
     """Return True when the on-disk context cache must not short-circuit probing.
 
@@ -2056,6 +2079,7 @@ def get_model_context_length(
 
     Resolution order:
     0. Explicit config override (model.context_length or custom_providers per-model)
+    0c. Endpoint-scoped metadata for models validated on one multiplexed endpoint
     1. Persistent cache (previously discovered via probing).  Nous URLs
        bypass the cache here so step 5b can always reconcile against
        the authoritative portal /v1/models response.
@@ -2125,11 +2149,29 @@ def get_model_context_length(
         except Exception:
             pass  # fall through to probing
 
+    # Malformed user-provided URLs (for example an unmatched IPv6 bracket)
+    # make urllib.parse raise. Context resolution should treat those as an
+    # unknown endpoint rather than crashing before the inference layer can
+    # report the configuration error itself.
+    if base_url:
+        try:
+            parsed_base_url = urlparse(_normalize_base_url(base_url))
+            _ = parsed_base_url.port
+        except ValueError:
+            base_url = ""
+
     # Normalise provider-prefixed model names (e.g. "local:model-name" →
     # "model-name") so cache lookups and server queries use the bare ID that
     # local servers actually know about.  Ollama "model:tag" colons are preserved.
     model = _strip_provider_prefix(model)
 
+    # Endpoint-scoped provider metadata. Keep this ahead of the persistent
+    # cache so a value learned for a multiplexed provider's other endpoint
+    # cannot override the endpoint where the model was actually validated.
+    endpoint_context = _endpoint_scoped_context_length(model, base_url)
+    if endpoint_context is not None:
+        return endpoint_context
+
     # 1. Check persistent cache (model+provider)
     # LM Studio is excluded — its loaded context length is transient (the
     # user can reload the model with a different context_length at any time

plugins/model-providers/kimi-coding/__init__.py

@@ -8,14 +8,55 @@ This module covers the chat_completions path (/v1 endpoint).
 """
 
 from typing import Any
+from urllib.parse import urlparse
 
 from providers import register_provider
 from providers.base import OMIT_TEMPERATURE, ProviderProfile
 
 
+def _is_confirmed_kimi_coding_url(base_url: str) -> bool:
+    """Return True only for Kimi Code's canonical HTTPS API surfaces."""
+    try:
+        parsed = urlparse(base_url)
+        port = parsed.port
+    except ValueError:
+        return False
+    return (
+        parsed.scheme.lower() == "https"
+        and (parsed.hostname or "").lower() == "api.kimi.com"
+        and port in (None, 443)
+        and parsed.username is None
+        and parsed.password is None
+        and parsed.path.rstrip("/") in {"/coding", "/coding/v1"}
+        and not parsed.query
+        and not parsed.fragment
+    )
+
+
 class KimiProfile(ProviderProfile):
     """Kimi/Moonshot — temperature omitted, thinking xor reasoning_effort."""
 
+    def fetch_models(
+        self,
+        *,
+        api_key: str | None = None,
+        base_url: str | None = None,
+        timeout: float = 8.0,
+    ) -> list[str] | None:
+        """Use Kimi Code's OpenAI-compatible surface for model discovery."""
+        effective_base = (base_url or self.base_url or "").rstrip("/")
+        confirmed_coding_endpoint = _is_confirmed_kimi_coding_url(effective_base)
+        if confirmed_coding_endpoint and urlparse(effective_base).path.rstrip("/") == "/coding":
+            effective_base += "/v1"
+        models = super().fetch_models(
+            api_key=api_key,
+            base_url=effective_base or None,
+            timeout=timeout,
+        )
+        if models is None or confirmed_coding_endpoint:
+            return models
+        return [model for model in models if model.strip().lower() != "k3"]
+
     def build_api_kwargs_extras(
         self, *, reasoning_config: dict | None = None, **context
     ) -> tuple[dict[str, Any], dict[str, Any]]:

tests/agent/test_model_metadata.py

@@ -174,6 +174,40 @@ class TestEstimateRequestTokensRough:
 # =========================================================================
 
 class TestDefaultContextLengths:
+    def test_k3_context_is_scoped_to_confirmed_coding_endpoint(self):
+        """K3's 1 Mi context must not leak to unverified Moonshot endpoints."""
+        with patch("agent.model_metadata.get_cached_context_length", return_value=None), \
+             patch("agent.model_metadata.fetch_model_metadata", return_value={}), \
+             patch("agent.model_metadata.fetch_endpoint_model_metadata", return_value={}), \
+             patch("agent.model_metadata._query_ollama_api_show", return_value=None), \
+             patch("agent.models_dev.lookup_models_dev_context", return_value=None):
+            accepted_urls = (
+                "https://api.kimi.com/coding",
+                "https://API.KIMI.COM/coding/",
+                "https://api.kimi.com:443/coding",
+                "https://api.kimi.com/coding/v1",
+            )
+            rejected_urls = (
+                "http://api.kimi.com/coding",
+                "https://api.kimi.com:8443/coding",
+                "https://api.kimi.com/coding/../other",
+                "https://api.kimi.com/codingevil",
+                "https://example.invalid/coding",
+                "https://[api.kimi.com/coding",
+                "https://api.moonshot.ai/v1",
+                "https://api.moonshot.cn/v1",
+            )
+
+            for base_url in accepted_urls:
+                assert get_model_context_length(
+                    "k3", provider="kimi-coding", base_url=base_url
+                ) == 1_048_576
+
+            for base_url in rejected_urls:
+                assert get_model_context_length(
+                    "k3", provider="kimi-coding", base_url=base_url
+                ) != 1_048_576
+
     def test_grok_substring_matching(self):
         # Longest-first substring matching must resolve the real xAI model
         # IDs to the correct fallback entries without 128k probe-down.

tests/hermes_cli/test_models_dev_preferred_merge.py

@@ -117,6 +117,64 @@ class TestProviderModelIdsPreferred:
         # Curated-first order; curated newest (k2.7-code) stays ahead of live.
         assert out[:2] == ["kimi-k2.7-code", "kimi-k2.6"]
 
+    def test_k3_live_discovery_is_scoped_to_kimi_coding_endpoint(self):
+        """Coding keys discover K3; legacy Moonshot keys must not advertise it."""
+
+        class Response:
+            def __init__(self, body: bytes):
+                self._body = body
+
+            def read(self):
+                return self._body
+
+            def __enter__(self):
+                return self
+
+            def __exit__(self, *_args):
+                return False
+
+        def fake_open(req, **_kwargs):
+            if req.full_url == "https://api.kimi.com/coding/v1/models":
+                return Response(b'{"data":[{"id":"k3"}]}')
+            if req.full_url == "https://api.moonshot.ai/v1/models":
+                return Response(b'{"data":[{"id":"K3"},{"id":"kimi-k2.6"}]}')
+            if req.full_url == "https://example.invalid/v1/models":
+                return Response(b'{"data":[{"id":"k3"},{"id":"kimi-k2.6"}]}')
+            raise AssertionError(f"unexpected Kimi models URL: {req.full_url}")
+
+        with patch("hermes_cli.urllib_security.open_credentialed_url", side_effect=fake_open):
+            with patch(
+                "hermes_cli.auth.resolve_api_key_provider_credentials",
+                return_value={
+                    "api_key": "sk-kimi-test",
+                    "base_url": "https://api.kimi.com/coding",
+                },
+            ):
+                coding_models = provider_model_ids("kimi-coding")
+
+            with patch(
+                "hermes_cli.auth.resolve_api_key_provider_credentials",
+                return_value={
+                    "api_key": "legacy-test",
+                    "base_url": "https://api.moonshot.ai/v1",
+                },
+            ):
+                legacy_models = provider_model_ids("kimi-coding")
+
+            with patch(
+                "hermes_cli.auth.resolve_api_key_provider_credentials",
+                return_value={
+                    "api_key": "custom-test",
+                    "base_url": "https://example.invalid/v1",
+                },
+            ):
+                custom_models = provider_model_ids("kimi-coding")
+
+        assert "k3" in coding_models
+        assert coding_models[0] == "kimi-k2.7-code"
+        assert all(model.lower() != "k3" for model in legacy_models)
+        assert all(model.lower() != "k3" for model in custom_models)
+
     def test_kimi_setup_flow_uses_same_coding_plan_catalog(self):
         """The setup wizard must not carry a stale duplicate Kimi model list."""
         from hermes_cli.model_setup_flows import _model_flow_kimi

tests/plugins/model_providers/test_kimi_profile.py

@@ -103,6 +103,26 @@ class TestKimiReasoningWireShape:
         assert not ("thinking" in extra_body and "reasoning_effort" in top_level)
 
 
+class TestKimiModelDiscovery:
+    def test_malformed_base_url_is_unconfirmed_and_filters_k3(self, kimi_profile):
+        """Malformed user URLs must fall through safely, never authorize K3."""
+        from unittest.mock import patch
+
+        from providers.base import ProviderProfile
+
+        with patch.object(
+            ProviderProfile,
+            "fetch_models",
+            return_value=["k3", "kimi-k2.6"],
+        ):
+            models = kimi_profile.fetch_models(
+                api_key="test-key",
+                base_url="https://[api.kimi.com/coding",
+            )
+
+        assert models == ["kimi-k2.6"]
+
+
 class TestKimiFullKwargsIntegration:
     """The transport's full kwargs carry at most one reasoning knob."""