hermes-agent/tests/agent/test_models_dev.py

"""Tests for agent.models_dev — models.dev registry integration."""
import threading
import time
from concurrent.futures import ThreadPoolExecutor
from unittest.mock import patch, MagicMock

import pytest

from agent.models_dev import (
    PROVIDER_TO_MODELS_DEV,
    _extract_context,
    fetch_models_dev,
    get_model_capabilities,
    get_provider_info,
    lookup_models_dev_context,
)


SAMPLE_REGISTRY = {
    "anthropic": {
        "id": "anthropic",
        "name": "Anthropic",
        "models": {
            "claude-opus-4-6": {
                "id": "claude-opus-4-6",
                "limit": {"context": 1000000, "output": 128000},
            },
            "claude-sonnet-4-6": {
                "id": "claude-sonnet-4-6",
                "limit": {"context": 1000000, "output": 64000},
            },
            "claude-sonnet-4-0": {
                "id": "claude-sonnet-4-0",
                "limit": {"context": 200000, "output": 64000},
            },
        },
    },
    "github-copilot": {
        "id": "github-copilot",
        "name": "GitHub Copilot",
        "models": {
            "claude-opus-4.6": {
                "id": "claude-opus-4.6",
                "limit": {"context": 128000, "output": 32000},
            },
        },
    },
    "xai": {
        "id": "xai",
        "name": "xAI",
        "models": {
            "grok-build-0.1": {
                "id": "grok-build-0.1",
                "limit": {"context": 256000, "output": 64000},
            },
        },
    },
    "kilo": {
        "id": "kilo",
        "name": "Kilo Gateway",
        "models": {
            "anthropic/claude-sonnet-4.6": {
                "id": "anthropic/claude-sonnet-4.6",
                "limit": {"context": 1000000, "output": 128000},
            },
        },
    },
    "deepseek": {
        "id": "deepseek",
        "name": "DeepSeek",
        "models": {
            "deepseek-chat": {
                "id": "deepseek-chat",
                "limit": {"context": 128000, "output": 8192},
            },
        },
    },
    "audio-only": {
        "id": "audio-only",
        "models": {
            "tts-model": {
                "id": "tts-model",
                "limit": {"context": 0, "output": 0},
            },
        },
    },
}


class TestProviderMapping:
    def test_xai_oauth_uses_xai_catalog(self):
        assert PROVIDER_TO_MODELS_DEV["xai"] == "xai"
        assert PROVIDER_TO_MODELS_DEV["xai-oauth"] == "xai"

    def test_unmapped_provider_not_in_dict(self):
        assert "nous" not in PROVIDER_TO_MODELS_DEV

    def test_openai_codex_mapped_to_openai(self):
        assert PROVIDER_TO_MODELS_DEV["openai"] == "openai"
        assert PROVIDER_TO_MODELS_DEV["openai-codex"] == "openai"


class TestExtractContext:
    def test_valid_entry(self):
        assert _extract_context({"limit": {"context": 128000}}) == 128000

    def test_zero_context_returns_none(self):
        assert _extract_context({"limit": {"context": 0}}) is None

    def test_missing_limit_returns_none(self):
        assert _extract_context({"id": "test"}) is None

    def test_missing_context_returns_none(self):
        assert _extract_context({"limit": {"output": 8192}}) is None

    def test_non_dict_returns_none(self):
        assert _extract_context("not a dict") is None

    def test_float_context_coerced_to_int(self):
        assert _extract_context({"limit": {"context": 131072.0}}) == 131072


class TestLookupModelsDevContext:
    @patch("agent.models_dev.fetch_models_dev")
    def test_exact_match(self, mock_fetch):
        mock_fetch.return_value = SAMPLE_REGISTRY
        assert lookup_models_dev_context("anthropic", "claude-opus-4-6") == 1000000

    @patch("agent.models_dev.fetch_models_dev")
    def test_case_insensitive_match(self, mock_fetch):
        mock_fetch.return_value = SAMPLE_REGISTRY
        assert lookup_models_dev_context("anthropic", "Claude-Opus-4-6") == 1000000

    @patch("agent.models_dev.fetch_models_dev")
    def test_provider_not_mapped(self, mock_fetch):
        mock_fetch.return_value = SAMPLE_REGISTRY
        assert lookup_models_dev_context("nous", "some-model") is None

    @patch("agent.models_dev.fetch_models_dev")
    def test_model_not_found(self, mock_fetch):
        mock_fetch.return_value = SAMPLE_REGISTRY
        assert lookup_models_dev_context("anthropic", "nonexistent-model") is None

    @patch("agent.models_dev.fetch_models_dev")
    def test_provider_aware_context(self, mock_fetch):
        """Same model, different context per provider."""
        mock_fetch.return_value = SAMPLE_REGISTRY
        # Anthropic direct: 1M
        assert lookup_models_dev_context("anthropic", "claude-opus-4-6") == 1000000
        # GitHub Copilot: only 128K for same model
        assert lookup_models_dev_context("copilot", "claude-opus-4.6") == 128000

    @patch("agent.models_dev.fetch_models_dev")
    def test_xai_oauth_resolves_xai_context(self, mock_fetch):
        """xAI OAuth is an auth path, not a separate model catalog."""
        mock_fetch.return_value = SAMPLE_REGISTRY
        assert lookup_models_dev_context("xai-oauth", "grok-build-0.1") == 256000

    @patch("agent.models_dev.fetch_models_dev")
    def test_zero_context_filtered(self, mock_fetch):
        mock_fetch.return_value = SAMPLE_REGISTRY
        # audio-only is not a mapped provider, but test the filtering directly
        data = SAMPLE_REGISTRY["audio-only"]["models"]["tts-model"]
        assert _extract_context(data) is None

    @patch("agent.models_dev.fetch_models_dev")
    def test_empty_registry(self, mock_fetch):
        mock_fetch.return_value = {}
        assert lookup_models_dev_context("anthropic", "claude-opus-4-6") is None


class TestFetchModelsDev:
    @pytest.fixture(autouse=True)
    def _reset_fetch_state(self):
        import agent.models_dev as md

        md._models_dev_cache = {}
        md._models_dev_cache_time = 0
        md._models_dev_retry_after = 0
        md._models_dev_refresh_in_flight = False
        yield
        md._models_dev_cache = {}
        md._models_dev_cache_time = 0
        md._models_dev_retry_after = 0
        md._models_dev_refresh_in_flight = False

    @patch("agent.models_dev.requests.get")
    def test_fetch_success(self, mock_get):
        mock_resp = MagicMock()
        mock_resp.status_code = 200
        mock_resp.json.return_value = SAMPLE_REGISTRY
        mock_resp.raise_for_status = MagicMock()
        mock_get.return_value = mock_resp

        # Clear caches
        import agent.models_dev as md
        md._models_dev_cache = {}
        md._models_dev_cache_time = 0

        with patch.object(md, "_save_disk_cache"):
            result = fetch_models_dev(force_refresh=True)

        assert "anthropic" in result
        assert len(result) == len(SAMPLE_REGISTRY)

    @patch("agent.models_dev.requests.get")
    def test_fetch_failure_returns_stale_cache(self, mock_get):
        mock_get.side_effect = Exception("network error")

        import agent.models_dev as md
        md._models_dev_cache = SAMPLE_REGISTRY
        md._models_dev_cache_time = 0  # expired

        with patch.object(md, "_load_disk_cache", return_value=SAMPLE_REGISTRY):
            result = fetch_models_dev(force_refresh=True)

        assert "anthropic" in result

    @patch("agent.models_dev.requests.get")
    def test_in_memory_cache_used(self, mock_get):
        import agent.models_dev as md
        import time
        md._models_dev_cache = SAMPLE_REGISTRY
        md._models_dev_cache_time = time.time()  # fresh

        result = fetch_models_dev()
        mock_get.assert_not_called()
        assert result == SAMPLE_REGISTRY

    @patch("agent.models_dev.requests.get")
    def test_stale_in_memory_cache_returns_without_foreground_network(self, mock_get):
        """Expired in-memory data should not block foreground resolution."""
        import agent.models_dev as md
        md._models_dev_cache = SAMPLE_REGISTRY
        md._models_dev_cache_time = 0

        with patch.object(md, "_start_background_refresh_models_dev") as mock_refresh:
            result = fetch_models_dev()

        mock_get.assert_not_called()
        mock_refresh.assert_called_once()
        assert result == SAMPLE_REGISTRY

    @patch("agent.models_dev.requests.get")
    def test_fresh_disk_cache_skips_network(self, mock_get):
        """When in-mem cache is empty but disk cache exists and is fresh by
        mtime (< TTL), fetch_models_dev returns disk data without ever
        making the network call.

        This is the cold-start fast path: every fresh process previously
        paid ~500 ms re-fetching a registry that was already on disk
        from an earlier run.
        """
        import agent.models_dev as md
        # Empty in-mem cache so stage 1 doesn't short-circuit.
        md._models_dev_cache = {}
        md._models_dev_cache_time = 0

        with patch.object(md, "_disk_cache_age_seconds", return_value=60.0), \
             patch.object(md, "_load_disk_cache", return_value=SAMPLE_REGISTRY):
            result = fetch_models_dev()

        # The whole point: no network call.
        mock_get.assert_not_called()
        assert "anthropic" in result
        # In-mem cache populated so subsequent calls within the same
        # process stay on stage 1.
        assert md._models_dev_cache == SAMPLE_REGISTRY

    @patch("agent.models_dev.requests.get")
    def test_stale_disk_cache_returns_without_foreground_network(self, mock_get):
        """#35838: stale disk cache should not wait on models.dev timeout."""
        import agent.models_dev as md
        md._models_dev_cache = {}
        md._models_dev_cache_time = 0

        with patch.object(md, "_disk_cache_age_seconds",
                          return_value=md._MODELS_DEV_CACHE_TTL + 60), \
             patch.object(md, "_load_disk_cache", return_value=SAMPLE_REGISTRY), \
             patch.object(md, "_start_background_refresh_models_dev") as mock_refresh:
            result = fetch_models_dev()

        mock_get.assert_not_called()
        mock_refresh.assert_called_once()
        assert "anthropic" in result

    @patch("agent.models_dev.requests.get")
    def test_force_refresh_skips_disk_cache(self, mock_get):
        """force_refresh=True bypasses BOTH the in-mem cache AND the
        disk-cache fast path. Used by ``hermes config refresh`` and
        anywhere else the user explicitly asked for fresh data.
        """
        import agent.models_dev as md
        md._models_dev_cache = {}
        md._models_dev_cache_time = 0

        mock_resp = MagicMock()
        mock_resp.status_code = 200
        mock_resp.json.return_value = SAMPLE_REGISTRY
        mock_resp.raise_for_status = MagicMock()
        mock_get.return_value = mock_resp

        # Disk cache is fresh, but force_refresh must override it.
        with patch.object(md, "_disk_cache_age_seconds", return_value=60.0), \
             patch.object(md, "_load_disk_cache", return_value=SAMPLE_REGISTRY), \
             patch.object(md, "_save_disk_cache"):
            result = fetch_models_dev(force_refresh=True)

        mock_get.assert_called_once()
        assert "anthropic" in result

    @patch("agent.models_dev.requests.get")
    def test_missing_disk_cache_falls_through_to_network(self, mock_get):
        """If the disk cache file doesn't exist (first-ever run, or it
        was deleted), fall through cleanly to network."""
        import agent.models_dev as md
        md._models_dev_cache = {}
        md._models_dev_cache_time = 0

        mock_resp = MagicMock()
        mock_resp.status_code = 200
        mock_resp.json.return_value = SAMPLE_REGISTRY
        mock_resp.raise_for_status = MagicMock()
        mock_get.return_value = mock_resp

        with patch.object(md, "_disk_cache_age_seconds", return_value=None), \
             patch.object(md, "_save_disk_cache"):
            result = fetch_models_dev()

        mock_get.assert_called_once()
        assert "anthropic" in result

    @patch("agent.models_dev.requests.get")
    def test_stale_cache_failure_enters_backoff_and_suppresses_retry(self, mock_get):
        import agent.models_dev as md

        mock_get.side_effect = OSError("models.dev unreachable")
        md._models_dev_cache = SAMPLE_REGISTRY
        md._models_dev_cache_time = time.time() - md._MODELS_DEV_CACHE_TTL - 1

        with patch.object(
            md,
            "_disk_cache_age_seconds",
            return_value=md._MODELS_DEV_CACHE_TTL + 60,
        ), patch.object(md, "_load_disk_cache", return_value=SAMPLE_REGISTRY):
            first = fetch_models_dev()
            # Wait for the background refresh worker to finish so its
            # failure backoff is observable and requests.get stays patched.
            deadline = time.time() + 5
            while md._models_dev_refresh_in_flight and time.time() < deadline:
                time.sleep(0.01)

        assert first == SAMPLE_REGISTRY
        assert not md._models_dev_refresh_in_flight
        assert md._models_dev_retry_after > time.time()
        mock_get.assert_called_once()

        # A subsequent stale-cache hit inside the backoff window must not
        # spawn another refresh worker (in_flight is set synchronously
        # before the worker thread starts, so False proves no spawn).
        md._models_dev_cache_time = time.time() - md._MODELS_DEV_CACHE_TTL - 1
        second = fetch_models_dev()
        assert second == SAMPLE_REGISTRY
        assert not md._models_dev_refresh_in_flight
        mock_get.assert_called_once()

    @patch("agent.models_dev.requests.get")
    def test_missing_cache_failure_enters_backoff(self, mock_get):
        import agent.models_dev as md

        mock_get.side_effect = OSError("models.dev unreachable")
        with patch.object(md, "_disk_cache_age_seconds", return_value=None), patch.object(
            md, "_load_disk_cache", return_value={}
        ):
            first = fetch_models_dev()
            second = fetch_models_dev()

        assert first == {}
        assert second == {}
        assert md._models_dev_retry_after > time.time()
        mock_get.assert_called_once()

    @patch("agent.models_dev.requests.get")
    def test_concurrent_refreshes_share_one_network_request(self, mock_get):
        import agent.models_dev as md

        request_started = threading.Event()
        release_request = threading.Event()
        response = MagicMock()
        response.json.return_value = SAMPLE_REGISTRY

        def blocking_get(*_args, **_kwargs):
            request_started.set()
            assert release_request.wait(timeout=5)
            return response

        mock_get.side_effect = blocking_get
        with patch.object(md, "_disk_cache_age_seconds", return_value=None), patch.object(
            md, "_save_disk_cache"
        ), ThreadPoolExecutor(max_workers=6) as pool:
            futures = [pool.submit(fetch_models_dev) for _ in range(6)]
            assert request_started.wait(timeout=2)
            release_request.set()
            results = [future.result(timeout=5) for future in futures]

        assert results == [SAMPLE_REGISTRY] * 6
        mock_get.assert_called_once()

    @patch("agent.models_dev.requests.get")
    def test_force_refresh_bypasses_failure_backoff(self, mock_get):
        import agent.models_dev as md

        response = MagicMock()
        response.json.return_value = SAMPLE_REGISTRY
        mock_get.side_effect = [OSError("models.dev unreachable"), response]

        with patch.object(md, "_disk_cache_age_seconds", return_value=None), patch.object(
            md, "_load_disk_cache", return_value={}
        ), patch.object(md, "_save_disk_cache"):
            assert fetch_models_dev() == {}
            assert fetch_models_dev(force_refresh=True) == SAMPLE_REGISTRY

        assert mock_get.call_count == 2
        assert md._models_dev_retry_after == 0

    @pytest.mark.parametrize(
        ("cache", "cache_time", "disk_data", "expected"),
        [
            (SAMPLE_REGISTRY, lambda md: time.time(), {}, SAMPLE_REGISTRY),
            (
                SAMPLE_REGISTRY,
                lambda md: time.time() - md._MODELS_DEV_CACHE_TTL - 1,
                {},
                SAMPLE_REGISTRY,
            ),
            ({}, lambda _md: 0, {}, {}),
        ],
        ids=["fresh-memory", "stale-memory", "missing"],
    )
    @patch("agent.models_dev.requests.get")
    def test_network_disabled_never_fetches(
        self, mock_get, cache, cache_time, disk_data, expected
    ):
        import agent.models_dev as md

        md._models_dev_cache = cache
        md._models_dev_cache_time = cache_time(md)
        with patch.object(md, "_load_disk_cache", return_value=disk_data):
            result = fetch_models_dev(allow_network=False)

        assert result == expected
        mock_get.assert_not_called()

    @patch("agent.models_dev.requests.get")
    def test_network_disabled_loads_stale_disk_cache(self, mock_get):
        import agent.models_dev as md

        with patch.object(md, "_load_disk_cache", return_value=SAMPLE_REGISTRY):
            result = fetch_models_dev(allow_network=False)

        assert result == SAMPLE_REGISTRY
        mock_get.assert_not_called()

    @patch("agent.models_dev.fetch_models_dev", return_value=SAMPLE_REGISTRY)
    def test_provider_info_propagates_network_disabled(self, mock_fetch):
        info = get_provider_info("anthropic", allow_network=False)

        assert info is not None
        mock_fetch.assert_called_once_with(allow_network=False)

    @patch("agent.models_dev.fetch_models_dev", return_value=SAMPLE_REGISTRY)
    def test_provider_info_default_preserves_zero_argument_fetch(self, mock_fetch):
        """Default path must stay a zero-arg call: many test sites monkeypatch
        fetch_models_dev with zero-arg lambdas."""
        info = get_provider_info("anthropic")

        assert info is not None
        mock_fetch.assert_called_once_with()

    def test_provider_definition_propagates_network_disabled(self):
        from hermes_cli.providers import get_provider

        with patch(
            "agent.models_dev.get_provider_info", return_value=None
        ) as mock_provider_info:
            get_provider("anthropic", allow_network=False)

        mock_provider_info.assert_called_once_with(
            "anthropic", allow_network=False
        )

    def test_default_route_lookup_is_cache_only(self):
        from agent.agent_init import _provider_default_routes

        with patch("hermes_cli.providers.get_provider", return_value=None) as mock_get:
            _provider_default_routes("anthropic")

        mock_get.assert_called_once_with("anthropic", allow_network=False)


# ---------------------------------------------------------------------------
# get_model_capabilities — vision via modalities.input
# ---------------------------------------------------------------------------


CAPS_REGISTRY = {
    "google": {
        "id": "google",
        "models": {
            "gemma-4-31b-it": {
                "id": "gemma-4-31b-it",
                "attachment": False,
                "tool_call": True,
                "modalities": {"input": ["text", "image"]},
                "limit": {"context": 128000, "output": 8192},
            },
            "gemma-3-1b": {
                "id": "gemma-3-1b",
                "tool_call": True,
                "limit": {"context": 32000, "output": 8192},
            },
            "text-only-with-stale-attachment": {
                "id": "text-only-with-stale-attachment",
                "attachment": True,
                "tool_call": True,
                "modalities": {"input": ["text"]},
                "limit": {"context": 128000, "output": 8192},
            },
        },
    },
    "anthropic": {
        "id": "anthropic",
        "models": {
            "claude-sonnet-4": {
                "id": "claude-sonnet-4",
                "attachment": True,
                "tool_call": True,
                "limit": {"context": 200000, "output": 64000},
            },
        },
    },
}


class TestGetModelCapabilities:
    """Tests for get_model_capabilities vision detection."""

    def test_vision_from_attachment_flag(self):
        """Models with attachment=True and no modalities should report supports_vision=True."""
        with patch("agent.models_dev.fetch_models_dev", return_value=CAPS_REGISTRY):
            caps = get_model_capabilities("anthropic", "claude-sonnet-4")
        assert caps is not None
        assert caps.supports_vision is True

    def test_vision_from_modalities_input_image(self):
        """Models with 'image' in modalities.input but attachment=False should
        still report supports_vision=True (the core fix in this PR)."""
        with patch("agent.models_dev.fetch_models_dev", return_value=CAPS_REGISTRY):
            caps = get_model_capabilities("google", "gemma-4-31b-it")
        assert caps is not None
        assert caps.supports_vision is True

    def test_text_only_modalities_override_stale_attachment_flag(self):
        """Text-only modalities must win over stale attachment=True metadata."""
        with patch("agent.models_dev.fetch_models_dev", return_value=CAPS_REGISTRY):
            caps = get_model_capabilities("google", "text-only-with-stale-attachment")
        assert caps is not None
        assert caps.supports_vision is False

    def test_no_vision_without_attachment_or_modalities(self):
        """Models with neither attachment nor image modality should be non-vision."""
        with patch("agent.models_dev.fetch_models_dev", return_value=CAPS_REGISTRY):
            caps = get_model_capabilities("google", "gemma-3-1b")
        assert caps is not None
        assert caps.supports_vision is False

    def test_modalities_non_dict_handled(self):
        """Non-dict modalities field should not crash."""
        registry = {
            "google": {"id": "google", "models": {
                "weird-model": {
                    "id": "weird-model",
                    "modalities": "text",  # not a dict
                    "limit": {"context": 200000, "output": 8192},
                },
            }},
        }
        with patch("agent.models_dev.fetch_models_dev", return_value=registry):
            caps = get_model_capabilities("gemini", "weird-model")
        assert caps is not None
        assert caps.supports_vision is False

    def test_model_not_found_returns_none(self):
        """Unknown model should return None."""
        with patch("agent.models_dev.fetch_models_dev", return_value=CAPS_REGISTRY):
            caps = get_model_capabilities("anthropic", "nonexistent-model")
        assert caps is None