"""Tests for agent/model_metadata.py — token estimation, context lengths, probing, caching, and error parsing. Coverage levels: Token estimation — concrete value assertions, edge cases Context length lookup — resolution order, fuzzy match, cache priority API metadata fetch — caching, TTL, canonical slugs, stale fallback Probe tiers — descending, boundaries, extreme inputs Error parsing — OpenAI, Ollama, Anthropic, edge cases Persistent cache — save/load, corruption, update, provider isolation """ import time import pytest import hermes_yaml as yaml from unittest.mock import patch, MagicMock from agent.model_metadata import ( CONTEXT_PROBE_TIERS, DEFAULT_CONTEXT_LENGTHS, DEFAULT_FALLBACK_CONTEXT, _strip_provider_prefix, estimate_tokens_rough, estimate_messages_tokens_rough, estimate_request_tokens_rough, get_model_context_length, get_next_probe_tier, get_cached_context_length, parse_context_limit_from_error, save_context_length, fetch_model_metadata, _MODEL_CACHE_TTL, ) def _codex_jwt(subject: str) -> str: """JWT-shaped test stand-in: the live-probe path gates on the token parsing as a JWT (a gateway key is not a ChatGPT credential and must stay off chatgpt.com, #121486); the signature itself is never verified client-side.""" import base64 as _b64 import json as _json def _enc(raw: bytes) -> str: return _b64.urlsafe_b64encode(raw).rstrip(b"=").decode() header = _enc(b'{"alg":"RS256"}') payload = _enc(_json.dumps({"sub": subject}).encode()) return f"{header}.{payload}.test-signature" # ========================================================================= # Token estimation # ========================================================================= class TestEstimateTokensRough: def test_empty_string(self): assert estimate_tokens_rough("") == 0 class TestEstimateMessagesTokensRough: def test_tool_call_message(self): """Tool call messages with no 'content' key still contribute tokens.""" msg = {"role": "assistant", "content": None, "tool_calls": [{"id": "1", "function": {"name": "terminal", "arguments": "{}"}}]} result = estimate_messages_tokens_rough([msg]) assert result > 0 def test_persistence_timestamp_does_not_change_estimate(self): """Durability metadata must not create artificial context pressure.""" msg = { "role": "assistant", "content": "done", "tool_calls": [ { "id": "call-1", "function": {"name": "terminal", "arguments": "{}"}, } ], } stamped = {**msg, "timestamp": 1_781_976_577.123456} assert estimate_messages_tokens_rough([stamped]) == ( estimate_messages_tokens_rough([msg]) ) def test_display_only_fields_do_not_change_estimate(self): """An edit row's inline_diff rides display_metadata, which the request builder strips; pricing it would compact early and break the prompt cache.""" wire = {"role": "tool", "tool_call_id": "call-1", "name": "write_file", "content": '{"bytes_written": 18000}'} rich = {**wire, "_row_id": 42, "display_kind": "tool_result", "display_metadata": {"tool_result_metadata": {"inline_diff": "\x1b[32m+ line\x1b[0m\n" * 600}}} assert estimate_messages_tokens_rough([rich]) == estimate_messages_tokens_rough([wire]) assert estimate_request_tokens_rough([rich]) == estimate_request_tokens_rough([wire]) def test_message_with_list_content(self): """Vision messages with multimodal content arrays. Image parts are counted at a flat ~1500-token rate per image rather than counting the base64 char length, so a tiny stub payload still registers as full image cost. """ msg = {"role": "user", "content": [ {"type": "text", "text": "describe"}, {"type": "image_url", "image_url": {"url": "data:image/png;base64,AAAA"}} ]} result = estimate_messages_tokens_rough([msg]) # Flat cost = 1500 per image plus the small text overhead. Allow # a small band so this isn't a change-detector for the exact # string representation. assert 1500 <= result < 2000 def test_api_content_substitutes_for_content_not_added_to_it(self): """``api_content`` replaces ``content`` on the wire, so count one. ``turn_context.substitute_api_content()`` pops the sidecar and overwrites ``content`` at every API-bound build site. Counting both doubled the estimate for any message carrying a sidecar. """ body = "cached prompt bytes " * 2000 wire_shape = {"role": "user", "content": body} persisted_shape = {"role": "user", "content": body, "api_content": body} assert estimate_messages_tokens_rough([persisted_shape]) == \ estimate_messages_tokens_rough([wire_shape]) def test_api_content_is_counted_when_it_differs_from_content(self): """The sidecar is what's sent, so its size is the one that matters.""" big_sidecar = "cached prompt bytes " * 2000 msg = {"role": "user", "content": "short", "api_content": big_sidecar} result = estimate_messages_tokens_rough([msg]) # Lower bound: fails if the sidecar were dropped rather than # substituted (which would undercount the real request). assert result >= (len(big_sidecar) // 4) * 0.9 def test_non_string_api_content_does_not_displace_content(self): """Only a sidecar shape the wire actually substitutes may displace content. ``substitute_api_content()`` overwrites ``content`` only for a non-empty STRING sidecar on a user/assistant row; every other shape is popped and discarded, leaving the clean ``content`` on the wire. The shadow must mirror that guard — substituting unconditionally would drop the real content from the estimate and UNDERcount, which is the dangerous direction (compaction fires too late and the turn dies on a hard context error). """ body = "clean stored content " * 2000 baseline = estimate_messages_tokens_rough([{"role": "user", "content": body}]) for bad_sidecar in (None, "", 42, ["not", "a", "string"]): msg = {"role": "user", "content": body, "api_content": bad_sidecar} assert estimate_messages_tokens_rough([msg]) >= baseline, bad_sidecar # Same for a role the substitution never applies to. tool_row = {"role": "tool", "content": body, "api_content": "ignored"} assert estimate_messages_tokens_rough([tool_row]) >= baseline def test_image_stripping_survives_shadow_extraction(self): """Non-regression for the ``_wire_message_shadow()`` extraction. Both estimator helpers now share one shadow builder; this pins the flat per-image accounting that the extraction moved, independent of the ``api_content`` fix (a valid sidecar is a string, so it cannot carry an image list). """ import base64 import os payload = "data:image/png;base64," + base64.b64encode(os.urandom(300_000)).decode() msg = {"role": "user", "content": [{"type": "image_url", "image_url": {"url": payload}}]} # Raw base64 would be ~100K tokens; the flat per-image model is ~1.5K. assert estimate_messages_tokens_rough([msg]) < 5_000 class TestResponsesItemImageAccounting: """Responses ``function_call_output`` items carry tool-result images under ``output`` (the converter moves chat ``content`` there); the estimator must price them with the flat per-image model, never as base64 text (#108320).""" def test_function_call_output_image_matches_chat_estimate(self): """The carrier key alone (``content`` vs ``output``) must not change the accounting for the same image.""" import base64 import os payload = ( "data:image/png;base64," + base64.b64encode(os.urandom(100_000)).decode() ) chat = { "role": "user", "content": [{"type": "image_url", "image_url": {"url": payload}}], } responses_item = { "type": "function_call_output", "call_id": "call_1", "output": [{"type": "input_image", "image_url": payload}], } chat_est = estimate_messages_tokens_rough([chat]) responses_est = estimate_messages_tokens_rough([responses_item]) assert abs(chat_est - responses_est) < 200 def test_function_call_output_text_output_still_counted(self): """Plain-string ``output`` (the common tool-result shape) is unaffected.""" item = { "type": "function_call_output", "call_id": "call_1", "output": "plain tool result " * 100, } est = estimate_messages_tokens_rough([item]) assert est >= (len(item["output"]) // 4) * 0.9 class TestEstimateRequestTokensRough: def test_tools_cache_is_bounded(self): # A long-lived process builds many transient tool lists; the cache must # not grow without bound. Feed more distinct lists than the cap and # confirm the cache never exceeds it. import agent.model_metadata as mm mm._TOOLS_TOKENS_CACHE.clear() cap = mm._TOOLS_TOKENS_CACHE_MAX # Keep references so ids are not recycled mid-loop, forcing distinct keys. held = [] for i in range(cap + 50): tools = [ { "type": "function", "function": { "name": f"tool_{i}", "description": "d", "parameters": {"type": "object"}, }, } ] held.append(tools) mm._estimate_tools_tokens_rough(tools) assert len(mm._TOOLS_TOKENS_CACHE) <= cap assert len(mm._TOOLS_TOKENS_CACHE) == cap # ========================================================================= # Default context lengths # ========================================================================= class TestDefaultContextLengths: def test_nvidia_deepseek_v4_pro_context_is_endpoint_scoped(self): """NVIDIA's 262K NIM window must not lower DeepSeek V4 globally.""" 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://integrate.api.nvidia.com/v1", "https://INTEGRATE.API.NVIDIA.COM/v1/", "https://integrate.api.nvidia.com:443/v1", ) rejected_urls = ( "http://integrate.api.nvidia.com/v1", "https://integrate.api.nvidia.com:8443/v1", "https://integrate.api.nvidia.com/v1/other", "https://integrate.api.nvidia.com/v1?route=other", "https://example.invalid/v1", "https://api.deepseek.com/v1", "https://openrouter.ai/api/v1", ) for base_url in accepted_urls: assert get_model_context_length( "deepseek-ai/deepseek-v4-pro", provider="nvidia", base_url=base_url, ) == 262_144 for base_url in rejected_urls: assert get_model_context_length( "deepseek-ai/deepseek-v4-pro", provider="nvidia", base_url=base_url, ) == 1_000_000 def test_k3_context_is_scoped_to_confirmed_coding_endpoint(self): """The bare ``k3`` slug's 1 Mi context must not leak to unverified endpoints. The named ``kimi-k3`` / ``kimi-k3-cot`` slugs resolve to 1 Mi EVERYWHERE via DEFAULT_CONTEXT_LENGTHS — the window is a property of the model, served at 1M on api.moonshot.ai and api.moonshot.cn alike (verified against models.dev + OpenRouter live metadata). Only the bare ``k3`` slug, which exists solely on the Kimi Coding Plan endpoint, stays endpoint-scoped. """ 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: for model in ("k3", "kimi-k3", "kimi-k3-cot"): assert get_model_context_length( model, provider="kimi-coding", base_url=base_url ) == 1_048_576 for base_url in rejected_urls: # Bare slug: endpoint-scoped, must NOT leak off-endpoint. assert get_model_context_length( "k3", provider="kimi-coding", base_url=base_url ) != 1_048_576 # Named slugs: global DEFAULT_CONTEXT_LENGTHS entry applies # everywhere the model is actually named kimi-k3. for model in ("kimi-k3", "kimi-k3-cot"): assert get_model_context_length( model, provider="kimi-coding", base_url=base_url ) == 1_048_576 @staticmethod def _upstage_ctx(model): with patch("agent.model_metadata.get_cached_context_length", return_value=None), \ patch("agent.model_metadata._query_ollama_api_show", return_value=None), \ patch("agent.model_metadata.fetch_endpoint_model_metadata", return_value={}), \ patch("agent.model_metadata.fetch_model_metadata", return_value={}), \ patch("agent.models_dev.fetch_models_dev", return_value={}): return get_model_context_length(model, provider="upstage", base_url="https://api.upstage.ai/v1") def test_upstage_solar_ids_match_legacy_keys_only_on_an_id_boundary(self): """Upstage /v1/models has no context field, so the table decides. ``solar-mini`` must not claim ``solar-mini4`` (its 32K is below MINIMUM_CONTEXT_LENGTH, so the agent refused to start); Solar ids without a legacy key get the Solar family window, not the 256K fallback.""" from agent.model_metadata import MINIMUM_CONTEXT_LENGTH, _longest_key_match family = DEFAULT_CONTEXT_LENGTHS["solar-"] assert family > DEFAULT_FALLBACK_CONTEXT for model in ("solar-mini4", "solar-mini4-preview", "upstage/solar-mini4", "solar-pro4", "solar-pro4-260806", "solar-pro4-quant", "solar-mini12", "solar-foo"): assert self._upstage_ctx(model) == family, model # The gateway labels a table miss as "default"; it must agree with the resolver. assert _longest_key_match(DEFAULT_CONTEXT_LENGTHS, model.lower())[1] == family, model # Legacy ids keep their own (smaller) windows: dated, org-prefixed, aggregator-hyphenated # (``solar-pro-3``) and quant/variant-suffixed ids included. for variant, bare in (("solar-mini-250422", "solar-mini"), ("upstage/solar-mini", "solar-mini"), ("solar-pro2-251215", "solar-pro2"), ("solar-pro3-260323", "solar-pro3"), ("upstage/solar-pro-3", "solar-pro3"), ("solar-pro3.1", "solar-pro3"), ("solar-open2@q4_k_m", "solar-open2")): assert self._upstage_ctx(variant) == self._upstage_ctx(bare), variant assert self._upstage_ctx("solar-mini") < MINIMUM_CONTEXT_LENGTH assert self._upstage_ctx("solar-pro3") < family # Ids merely containing "solar", and open-weight Solar ids, are not Solar API lineups. for model in ("ft:solar-news-correction", "acme-solar-foo", "upstage/solar-10.7b-instruct"): assert self._upstage_ctx(model) != family, model def test_explicit_solar_key_beats_the_family_default(self): with patch.dict(DEFAULT_CONTEXT_LENGTHS, {"solar-foo": 300_000}): assert self._upstage_ctx("solar-foo") == 300_000 assert self._upstage_ctx("solar-foo-260101") == 300_000 assert self._upstage_ctx("solar-foo2") == DEFAULT_CONTEXT_LENGTHS["solar-"] def test_empty_model_uses_fallback_context(self): assert get_model_context_length("") == DEFAULT_FALLBACK_CONTEXT assert get_model_context_length(None) == DEFAULT_FALLBACK_CONTEXT # type: ignore[arg-type] def test_xai_oauth_grok_build_uses_xai_models_dev_context(self): """xAI OAuth should share the xAI provider metadata path. The xAI /v1/models endpoint does not currently include context fields for grok-build-0.1, so this guards against falling through to the generic "grok" 131k fallback when using OAuth credentials. """ registry = { "xai": { "models": { "grok-build-0.1": { "limit": {"context": 256000, "output": 64000}, }, }, }, } with patch("agent.model_metadata.get_cached_context_length", return_value=None), \ patch("agent.model_metadata._query_ollama_api_show", return_value=None), \ patch("agent.models_dev.fetch_models_dev", return_value=registry): assert get_model_context_length( "grok-build-0.1", provider="xai-oauth", base_url="https://api.x.ai/v1", api_key="oauth-token", ) == 256000 # ========================================================================= # Codex OAuth context-window resolution (provider="openai-codex") # ========================================================================= class TestCodexOAuthContextLength: """ChatGPT Codex OAuth context windows come from the authenticated /models catalogue and may differ from the static fallback table or the direct OpenAI API allocation. The fallback values below are conservative defaults used only when the live probe is unavailable. """ def setup_method(self): import agent.model_metadata as mm mm._codex_oauth_context_cache = {} mm._codex_oauth_max_context_cache = {} # the fallback branch reads it too def test_live_catalogue_cache_is_scoped_to_access_token(self): """Different OAuth tokens must not share entitlement-specific metadata.""" from agent import model_metadata as mm from agent.model_metadata import get_model_context_length first_response = MagicMock() first_response.status_code = 200 first_response.json.return_value = { "models": [{"slug": "gpt-5.5", "context_window": 272_000}] } second_response = MagicMock() second_response.status_code = 200 second_response.json.return_value = { "models": [{"slug": "gpt-5.5", "context_window": 372_000}] } with patch( "agent.model_metadata.model_metadata_http.get", side_effect=[first_response, second_response], ) as mock_get, patch("agent.model_metadata.save_context_length") as mock_save: first = get_model_context_length( "gpt-5.5", base_url="https://chatgpt.com/backend-api/codex", api_key=_codex_jwt("token-account-a"), provider="openai-codex", ) first_again = get_model_context_length( "gpt-5.5", base_url="https://chatgpt.com/backend-api/codex", api_key=_codex_jwt("token-account-a"), provider="openai-codex", ) second = get_model_context_length( "gpt-5.5", base_url="https://chatgpt.com/backend-api/codex", api_key=_codex_jwt("token-account-b"), provider="openai-codex", ) assert (first, first_again, second) == (272_000, 272_000, 372_000) assert mock_get.call_count == 2 assert mock_get.call_args_list[0].kwargs["headers"]["Authorization"] == f"Bearer {_codex_jwt('token-account-a')}" assert mock_get.call_args_list[1].kwargs["headers"]["Authorization"] == f"Bearer {_codex_jwt('token-account-b')}" assert mock_save.call_count == 2 assert all( "token-account" not in key for key in mm._codex_oauth_context_cache ) def test_probe_failure_falls_back_to_hardcoded(self): """If the probe fails (non-200 / network error), we still return the hardcoded 272k rather than leaking through to models.dev 1.05M.""" from agent.model_metadata import get_model_context_length fake_response = MagicMock() fake_response.status_code = 401 fake_response.json.return_value = {} with patch("agent.model_metadata.model_metadata_http.get", return_value=fake_response), \ patch("agent.model_metadata.get_cached_context_length", return_value=None), \ patch("agent.model_metadata.save_context_length"): ctx = get_model_context_length( model="gpt-5.5", base_url="https://chatgpt.com/backend-api/codex", api_key=_codex_jwt("expired-token"), provider="openai-codex", ) assert ctx == 272_000 def test_gateway_key_never_probes_the_direct_catalog(self): """A custom base's credential is the gateway's key, not a ChatGPT token: sending it to chatgpt.com leaks it to a service it does not belong to and can never answer (#121486). The probe must decline (falling back to the static table) instead of firing.""" import agent.model_metadata as mm mm._codex_oauth_context_cache = {} with patch("agent.model_metadata.model_metadata_http.get") as mock_get: live, fresh = mm._fetch_codex_oauth_context_lengths_with_source("gateway-pool-key") assert (live, fresh) == ({}, False) mock_get.assert_not_called() def test_custom_base_probes_its_own_catalog(self): """A JWT reached through a custom Codex base probes that base's ``/models``, never the hard-coded chatgpt.com host (#121486).""" import agent.model_metadata as mm mm._codex_oauth_context_cache = {} fake_response = MagicMock() fake_response.status_code = 200 fake_response.json.return_value = { "models": [{"slug": "gpt-5.5", "context_window": 272_000}] } with patch("agent.model_metadata.model_metadata_http.get", return_value=fake_response) as mock_get, \ patch("agent.model_metadata.get_cached_context_length", return_value=None), \ patch("agent.model_metadata.save_context_length"): ctx = get_model_context_length( model="gpt-5.5", base_url="https://codex-gw.example/backend-api/codex", api_key=_codex_jwt("acct"), provider="openai-codex", ) assert ctx == 272_000 assert mock_get.call_args.args[0].startswith( "https://codex-gw.example/backend-api/codex/models?client_version=") @pytest.mark.parametrize( "stale_context,live_context", [(272_000, 372_000), (372_000, 272_000)], ids=("expansion", "rollback"), ) def test_live_codex_context_replaces_stale_cache_in_both_directions( self, tmp_path, monkeypatch, stale_context, live_context ): """Authenticated metadata must replace stale disk values in either direction.""" from agent import model_metadata as mm cache_file = tmp_path / "context_length_cache.yaml" monkeypatch.setattr(mm, "_get_context_cache_path", lambda: cache_file) base_url = "https://chatgpt.com/backend-api/codex" stale_key = f"gpt-5.5@{base_url}" other_key = "other-model@https://api.openai.com/v1/" import hermes_yaml as _yaml cache_file.write_text(_yaml.safe_dump({"context_lengths": { stale_key: stale_context, other_key: 128_000, }})) fake_response = MagicMock() fake_response.status_code = 200 fake_response.json.return_value = { "models": [{"slug": "gpt-5.5", "context_window": live_context}] } # Exercise real persistence here: this test verifies that a live value # replaces the stale on-disk entry. Failure-path tests below mock the # writer because they assert that fallback values are not persisted. with patch("agent.model_metadata.model_metadata_http.get", return_value=fake_response) as mock_get: ctx = mm.get_model_context_length( model="gpt-5.5", base_url=base_url, api_key=_codex_jwt("fake-token"), provider="openai-codex", ) assert ctx == live_context mock_get.assert_called_once() remaining = _yaml.safe_load(cache_file.read_text(encoding="utf-8")).get( "context_lengths", {} ) assert remaining.get(stale_key) == live_context assert remaining.get(other_key) == 128_000 @pytest.mark.parametrize("slug", ["gpt-5.6-sol"]) def test_base_slug_keeps_advertised_272k(self, slug): """Base slugs (no ``-900k`` suffix) keep the advertised 272K — the cheaper default limit. The verified-above bump is opt-in only.""" from agent.model_metadata import get_model_context_length fake_response = MagicMock() fake_response.status_code = 200 fake_response.json.return_value = { "models": [{"slug": slug, "context_window": 272_000}] } import agent.model_metadata as mm mm._codex_oauth_context_cache = {} with patch("agent.model_metadata.model_metadata_http.get", return_value=fake_response), \ patch("agent.model_metadata.get_cached_context_length", return_value=None), \ patch("agent.model_metadata.save_context_length"): ctx = get_model_context_length( model=slug, base_url="https://chatgpt.com/backend-api/codex", api_key=_codex_jwt("fake-token"), provider="openai-codex", ) assert ctx == 272_000 def test_non_272k_advertisement_is_trusted_verbatim(self): """Any advertised value other than the known-stale 272,000 — higher or lower — is a real server-side change and must NOT be overridden, even for an explicit ``-900k`` opt-in variant.""" from agent.model_metadata import get_model_context_length for advertised in (372_000, 200_000, 1_050_000): fake_response = MagicMock() fake_response.status_code = 200 fake_response.json.return_value = { "models": [{"slug": "gpt-5.6-sol", "context_window": advertised}] } import agent.model_metadata as mm mm._codex_oauth_context_cache = {} with patch("agent.model_metadata.model_metadata_http.get", return_value=fake_response), \ patch("agent.model_metadata.get_cached_context_length", return_value=None), \ patch("agent.model_metadata.save_context_length"): ctx = get_model_context_length( model="gpt-5.6-sol-900k", base_url="https://chatgpt.com/backend-api/codex", api_key=_codex_jwt("fake-token"), provider="openai-codex", ) assert ctx == advertised, f"advertised {advertised} must be trusted" @pytest.mark.parametrize("base", ["gpt-5.6-luna", "gpt-6.1-sol"]) @pytest.mark.parametrize("catalog_max,expected", [(872_000, 872_000), (None, 900_000), (1_050_000, 900_000)]) def test_opted_in_variant_capped_at_live_catalog_max(self, base, catalog_max, expected): """An explicit ``-900k`` opt-in resolves to min(900K, catalog ``max_context_window``): gpt-5.6 advertises 272K with an 872K max (#105443); a catalog without the field or one above the live-verified cap keeps 900K. The 900K table entry is only the fallback.""" from agent.model_metadata import get_model_context_length item = {"slug": base, "context_window": 272_000} if catalog_max is not None: item["max_context_window"] = catalog_max fake_response = MagicMock() fake_response.status_code = 200 fake_response.json.return_value = {"models": [item]} with patch("agent.model_metadata.model_metadata_http.get", return_value=fake_response), \ patch("agent.model_metadata.get_cached_context_length", return_value=None), \ patch("agent.model_metadata.save_context_length"): ctx = get_model_context_length( model=f"{base}-900k", base_url="https://chatgpt.com/backend-api/codex", api_key=_codex_jwt("fake-token"), provider="openai-codex", ) assert ctx == expected def test_base_slug_keeps_advertised_ctx_even_with_catalog_max(self): """The catalogue max never leaks into the base slug: extended context is opt-in via the ``-900k`` alias only (#105443).""" from agent.model_metadata import get_model_context_length fake_response = MagicMock() fake_response.status_code = 200 fake_response.json.return_value = { "models": [{"slug": "gpt-5.6-sol", "context_window": 272_000, "max_context_window": 872_000}] } with patch("agent.model_metadata.model_metadata_http.get", return_value=fake_response), \ patch("agent.model_metadata.get_cached_context_length", return_value=None), \ patch("agent.model_metadata.save_context_length"): ctx = get_model_context_length( model="gpt-5.6-sol", base_url="https://chatgpt.com/backend-api/codex", api_key=_codex_jwt("fake-token"), provider="openai-codex", ) assert ctx == 272_000 def test_failed_refresh_past_ttl_keeps_the_observed_catalog_cap(self): """A refresh that fails after the context TTL must not lift a cap the same token and endpoint already published: no newer evidence, so the -900k alias stays at the old max.""" import agent.model_metadata as mm ok = MagicMock(status_code=200) ok.json.return_value = {"models": [{"slug": "GPT-5.6-Luna", "context_window": 272_000, "max_context_window": 600_000}]} route = dict(base_url="https://chatgpt.com/backend-api/codex", api_key=_codex_jwt("fake-token"), provider="openai-codex") with patch.object(mm, "_codex_oauth_context_cache", {}), patch.object(mm, "_codex_oauth_max_context_cache", {}), \ patch("agent.model_metadata.get_cached_context_length", return_value=None), \ patch("agent.model_metadata.save_context_length"): with patch("agent.model_metadata.model_metadata_http.get", return_value=ok): assert mm.get_model_context_length(model="gpt-5.6-luna-900k", **route) == 600_000 with patch("agent.model_metadata.model_metadata_http.get", side_effect=OSError("catalog down")), \ patch("agent.model_metadata.time.time", return_value=time.time() + mm._CODEX_OAUTH_CONTEXT_CACHE_TTL + 1): assert mm.get_model_context_length(model="gpt-5.6-luna-900k", **route) == 600_000 def test_a_probe_that_finds_no_catalog_is_not_repeated_within_the_negative_ttl(self): """Every context resolution (init, /status, aux, @-reference turns) used to re-probe a dead or non-Codex-shaped catalog; one miss is now remembered for the short negative TTL, then retried.""" import agent.model_metadata as mm calls = [] def down(url, **_kw): calls.append(url) raise OSError("catalog down") route = dict(base_url="https://chatgpt.com/backend-api/codex", api_key=_codex_jwt("fake-token"), provider="openai-codex") with patch.object(mm, "_codex_oauth_context_cache", {}), patch.object(mm, "_codex_oauth_max_context_cache", {}), \ patch("agent.model_metadata.model_metadata_http.get", side_effect=down), \ patch("agent.model_metadata.get_cached_context_length", return_value=None), \ patch("agent.model_metadata.save_context_length"): assert mm.get_model_context_length(model="gpt-5.6-sol", **route) == 272_000 per_probe = len(calls) mm.get_model_context_length(model="gpt-5.6-sol", **route) mm.get_model_context_length(model="gpt-5.6-sol-900k", **route) assert per_probe and len(calls) == per_probe with patch("agent.model_metadata.time.time", return_value=time.time() + mm._CODEX_OAUTH_CONTEXT_NEGATIVE_TTL + 1): mm.get_model_context_length(model="gpt-5.6-sol", **route) assert len(calls) == 2 * per_probe def test_catalog_max_is_published_before_the_context_entry(self): """A reader that sees a fresh context entry must already be able to see its cap, or a concurrent -900k resolution between the two writes skips the catalog max.""" import agent.model_metadata as mm max_cache: dict = {} class _ContextCache(dict): def __setitem__(self, key, value): assert key in max_cache, "context entry published before its max_context_window cap" super().__setitem__(key, value) ok = MagicMock(status_code=200) ok.json.return_value = {"models": [{"slug": "gpt-5.6-luna", "context_window": 272_000, "max_context_window": 872_000}]} with patch.object(mm, "_codex_oauth_context_cache", _ContextCache()), patch.object(mm, "_codex_oauth_max_context_cache", max_cache), \ patch("agent.model_metadata.model_metadata_http.get", return_value=ok): live, _fresh = mm._fetch_codex_oauth_context_lengths_with_source(_codex_jwt("fake-token")) assert live == {"gpt-5.6-luna": 272_000} @pytest.mark.parametrize("slug", ["gpt-5.6-sol-900k"]) def test_fallback_table_resolution_also_bumped(self, slug): """When the live probe fails, the 272K fallback-table value for an opted-in ``-900k`` variant is bumped the same way (same enforcement applies — the fallback lookup strips the suffix first).""" from agent.model_metadata import get_model_context_length fake_response = MagicMock() fake_response.status_code = 401 fake_response.json.return_value = {} with patch("agent.model_metadata.model_metadata_http.get", return_value=fake_response), \ patch("agent.model_metadata.get_cached_context_length", return_value=None), \ patch("agent.model_metadata.save_context_length"): ctx = get_model_context_length( model=slug, base_url="https://chatgpt.com/backend-api/codex", api_key=_codex_jwt("expired-token"), provider="openai-codex", ) assert ctx == 900_000 @pytest.mark.parametrize("slug", ["gpt-5.6-sol"]) def test_fallback_table_base_slug_stays_272k(self, slug): """Fallback-table resolution for BASE slugs stays at the advertised 272K — the opt-in rule applies on the offline path too.""" from agent.model_metadata import get_model_context_length fake_response = MagicMock() fake_response.status_code = 401 fake_response.json.return_value = {} with patch("agent.model_metadata.model_metadata_http.get", return_value=fake_response), \ patch("agent.model_metadata.get_cached_context_length", return_value=None), \ patch("agent.model_metadata.save_context_length"): ctx = get_model_context_length( model=slug, base_url="https://chatgpt.com/backend-api/codex", api_key=_codex_jwt("expired-token"), provider="openai-codex", ) assert ctx == 272_000 # Table-driven eligibility contract (#92797 review): one predicate # (is_codex_900k_base) drives picker synthesis, context resolution, # validation, and wire stripping — this table pins all of them. # (model_id, is_valid_variant, expected_ctx, expected_wire_model) _900K_TABLE = [ ("gpt-5.6-sol-900k", True, 900_000, "gpt-5.6-sol"), # dated snapshot of a routable 5.6 base ("gpt-5.6-sol-2026-07-09-900k", True, 900_000, "gpt-5.6-sol-2026-07-09"), # vendor-namespaced variant (display/aux callers) resolves too ("openai/gpt-5.6-sol-900k", True, 900_000, "openai/gpt-5.6-sol"), # -pro slugs are not routable on Codex OAuth: never a valid variant, # never stripped (fails honestly at the API instead) ("gpt-5.6-sol-pro-900k", False, 272_000, "gpt-5.6-sol-pro-900k"), # genuine 272K enforcers get no variant ("gpt-5.5-900k", False, 272_000, "gpt-5.5-900k"), # arbitrary future family descendants are not auto-eligible ("gpt-5.6-nova-900k", False, 272_000, "gpt-5.6-nova-900k"), ] @pytest.mark.parametrize("model_id,valid,expected_ctx,wire", _900K_TABLE) def test_900k_eligibility_table(self, model_id, valid, expected_ctx, wire): from agent.model_metadata import ( get_model_context_length, is_codex_context_variant, strip_codex_context_variant_suffix, ) assert is_codex_context_variant(model_id) is valid assert strip_codex_context_variant_suffix(model_id) == wire bare = model_id.rsplit("/", 1)[-1] catalog_slug = strip_codex_context_variant_suffix(bare) if catalog_slug.endswith("-900k"): # invalid alias — catalog advertises the underlying family slug catalog_slug = catalog_slug[: -len("-900k")] fake_response = MagicMock() fake_response.status_code = 200 fake_response.json.return_value = { "models": [{"slug": catalog_slug, "context_window": 272_000}] } import agent.model_metadata as mm mm._codex_oauth_context_cache = {} with patch("agent.model_metadata.model_metadata_http.get", return_value=fake_response), \ patch("agent.model_metadata.get_cached_context_length", return_value=None), \ patch("agent.model_metadata.save_context_length"): ctx = get_model_context_length( model=model_id, base_url="https://chatgpt.com/backend-api/codex", api_key=_codex_jwt("fake-token"), provider="openai-codex", ) assert ctx == expected_ctx # ========================================================================= # Custom endpoint model metadata # ========================================================================= def _streamed(response): """A ``model_metadata_http.stream`` result: a context manager yielding *response*.""" ctx = MagicMock() ctx.__enter__.return_value = response return ctx class TestFetchEndpointModelMetadata: def setup_method(self): import agent.model_metadata as mm mm._endpoint_model_metadata_cache.clear() mm._endpoint_model_metadata_cache_time.clear() @pytest.mark.parametrize("status_code", [401, 403]) def test_auth_failure_stops_after_first_candidate(self, status_code): import agent.model_metadata as mm response = MagicMock() response.status_code = status_code response.raise_for_status.side_effect = RuntimeError(str(status_code)) ctx = _streamed(response) with patch("agent.model_metadata.model_metadata_http.stream", return_value=ctx) as mock_stream: result = mm.fetch_endpoint_model_metadata("https://custom.example/v1") assert result == {} mock_stream.assert_called_once() response.raise_for_status.assert_not_called() response.json.assert_not_called() ctx.__exit__.assert_called_once() def test_auth_failure_empty_result_is_cached(self): import agent.model_metadata as mm response = MagicMock() response.status_code = 401 response.raise_for_status.side_effect = RuntimeError("401") ctx = _streamed(response) with patch("agent.model_metadata.model_metadata_http.stream", return_value=ctx) as mock_stream: first = mm.fetch_endpoint_model_metadata("https://custom.example/v1") second = mm.fetch_endpoint_model_metadata("https://custom.example/v1") assert first == second == {} mock_stream.assert_called_once() ctx.__exit__.assert_called_once() def test_not_found_still_tries_alternate_candidate(self): import agent.model_metadata as mm not_found = MagicMock() not_found.status_code = 404 not_found.raise_for_status.side_effect = RuntimeError("404") success = MagicMock() success.status_code = 200 success.json.return_value = { "data": [{"id": "test/model", "context_length": 32768}] } not_found_ctx, success_ctx = _streamed(not_found), _streamed(success) with patch( "agent.model_metadata.model_metadata_http.stream", side_effect=[not_found_ctx, success_ctx], ) as mock_stream: result = mm.fetch_endpoint_model_metadata("https://custom.example/v1") assert result["test/model"]["context_length"] == 32768 assert [call.args[0] for call in mock_stream.call_args_list] == [ "https://custom.example/v1/models", "https://custom.example/models", ] not_found.json.assert_not_called() not_found_ctx.__exit__.assert_called_once() success_ctx.__exit__.assert_called_once() def test_remote_probe_is_memoized_on_disk_across_processes(self, tmp_path, monkeypatch): """A fresh process (cleared in-memory cache) must answer from the disk memo within the TTL instead of re-probing the endpoint — the cost every one-shot Bot Mode DM hop paid on startup. Expired memos re-probe.""" import agent.model_metadata as mm monkeypatch.setattr( mm, "_get_endpoint_metadata_cache_path", lambda: tmp_path / "endpoint_model_metadata.json" ) success = MagicMock() success.status_code = 200 success.json.return_value = {"data": [{"id": "test/model", "context_length": 32768}]} with patch("agent.model_metadata.model_metadata_http.stream", return_value=_streamed(success)) as mock_stream: assert mm.fetch_endpoint_model_metadata("https://custom.example/v1")["test/model"]["context_length"] == 32768 # "New process": drop the in-memory cache only. mm._endpoint_model_metadata_cache.clear() mm._endpoint_model_metadata_cache_time.clear() assert mm.fetch_endpoint_model_metadata("https://custom.example/v1")["test/model"]["context_length"] == 32768 mock_stream.assert_called_once() # Past the TTL the memo is stale and the endpoint is probed again. mm._endpoint_model_metadata_cache.clear() mm._endpoint_model_metadata_cache_time.clear() with patch("agent.model_metadata.time.time", return_value=time.time() + mm._ENDPOINT_MODEL_CACHE_TTL + 1), patch( "agent.model_metadata.model_metadata_http.stream", return_value=_streamed(success) ) as mock_stream: mm.fetch_endpoint_model_metadata("https://custom.example/v1") mock_stream.assert_called_once() # ========================================================================= # Nous Portal context-window resolution (provider="nous") # ========================================================================= class TestNousPortalContextResolution: """Nous Portal /v1/models is authoritative for what Nous infra enforces and may diverge from the OpenRouter catalog. Invariants this class pins down: 1. Portal value wins over the OR fallback. 2. Portal-derived values are persisted to disk. 3. OR-fallback values are NEVER persisted — otherwise a single portal blip would freeze the wrong value in via step-1 cache short-circuit. 4. Pre-fix persistent-cache entries (seeded from the OR catalog) are bypassed at step 1 and overwritten once the portal responds. 5. Pre-fix persistent-cache entries SURVIVE on disk when the portal is unreachable — no opportunistic invalidation that loses the only value we have. """ def setup_method(self): import agent.model_metadata as mm mm._endpoint_model_metadata_cache.clear() mm._endpoint_model_metadata_cache_time.clear() @patch("agent.model_metadata.fetch_endpoint_model_metadata") def test_empty_model_never_fuzzy_matches_endpoint_catalog(self, mock_fetch): """An empty model name must not substring-match arbitrary catalog entries — '' is a substring of every key, so pre-fix it "matched" whatever the endpoint listed first (e.g. a 32K embedding model on the Nous portal) and poisoned the resolved context length.""" import agent.model_metadata as mm mock_fetch.return_value = { "voyageai/voyage-code-4": {"context_length": 32_000}, "x-ai/grok-4.6": {"context_length": 500_000}, } assert mm._resolve_endpoint_context_length( "", "https://inference-api.nousresearch.com/v1" ) is None # Non-empty names still fuzzy-match. assert mm._resolve_endpoint_context_length( "grok-4.6", "https://inference-api.nousresearch.com/v1" ) == 500_000 # Single-model endpoints still resolve even with an empty name. mock_fetch.return_value = {"only-model": {"context_length": 131_072}} assert mm._resolve_endpoint_context_length( "", "http://localhost:8080/v1" ) == 131_072 @patch("agent.model_metadata.fetch_endpoint_model_metadata") @patch("agent.model_metadata.fetch_model_metadata") def test_openrouter_fallback_is_not_persisted( self, mock_or, mock_portal, tmp_path, monkeypatch ): """When the portal can't resolve a model (network blip, auth glitch, model not yet listed) we fall back to the OR catalog so the agent keeps working — but we must NOT write the OR value to disk. Once cached on disk, step-1 short-circuits forever and the user is stuck with the wrong number until they manually clear the cache.""" import agent.model_metadata as mm cache_file = tmp_path / "context_length_cache.yaml" monkeypatch.setattr(mm, "_get_context_cache_path", lambda: cache_file) mock_portal.return_value = {} # portal unreachable / model unknown mock_or.return_value = { "qwen/qwen3.6-plus": {"context_length": 1_000_000}, } base_url = "https://inference-api.nousresearch.com/v1" ctx = mm.get_model_context_length( model="qwen3.6-plus", base_url=base_url, api_key="fake", provider="nous", ) assert ctx == 1_000_000, "OR fallback should still serve the request" assert not cache_file.exists() or not yaml.safe_load( cache_file.read_text(encoding="utf-8") ).get("context_lengths", {}), ( "OR-fallback values must NOT be persisted — a single portal blip " "would otherwise freeze the wrong value in via step-1 cache hit" ) @patch("agent.model_metadata.fetch_endpoint_model_metadata") @patch("agent.model_metadata.fetch_model_metadata") def test_stale_cache_is_bypassed_and_overwritten_by_portal( self, mock_or, mock_portal, tmp_path, monkeypatch ): """Users upgrading from pre-fix builds have ``qwen3.6-plus@…nous… = 1000000`` (OR-derived) sitting in their cache file. Step 1 must NOT short-circuit on that entry — step 5b reconciles against the portal and overwrites the persistent value with 262144.""" import agent.model_metadata as mm cache_file = tmp_path / "context_length_cache.yaml" monkeypatch.setattr(mm, "_get_context_cache_path", lambda: cache_file) base_url = "https://inference-api.nousresearch.com/v1" stale_key = f"qwen3.6-plus@{base_url}" other_key = "other-model@https://api.openai.com/v1" cache_file.write_text(yaml.safe_dump({"context_lengths": { stale_key: 1_000_000, # pre-fix OR-derived value other_key: 128_000, # unrelated, must survive }})) mock_portal.return_value = { "qwen3.6-plus": {"context_length": 262_144}, } mock_or.return_value = {} ctx = mm.get_model_context_length( model="qwen3.6-plus", base_url=base_url, api_key="fake", provider="nous", ) assert ctx == 262_144, ( f"Stale OR-derived cache entry should not have leaked through; got {ctx}" ) remaining = yaml.safe_load(cache_file.read_text(encoding="utf-8")).get( "context_lengths", {} ) assert remaining.get(stale_key) == 262_144, ( "Portal value should have overwritten the stale entry on disk" ) assert remaining.get(other_key) == 128_000, ( "Unrelated cache entries must not be touched" ) # ========================================================================= # get_model_context_length — resolution order # ========================================================================= class TestGetModelContextLength: @patch("agent.model_metadata.fetch_model_metadata") def test_known_model_from_api(self, mock_fetch): mock_fetch.return_value = { "test/model": {"context_length": 32000} } assert get_model_context_length("test/model") == 32000 @patch("agent.model_metadata.fetch_model_metadata") def test_api_missing_context_length_key(self, mock_fetch): """Model in API but without context_length → defaults to the top probe tier (currently 256K).""" mock_fetch.return_value = {"test/model": {"name": "Test"}} assert get_model_context_length("test/model") == CONTEXT_PROBE_TIERS[0] @patch("agent.model_metadata.fetch_model_metadata") def test_no_base_url_skips_cache(self, mock_fetch, tmp_path): """Without base_url, cache lookup is skipped.""" mock_fetch.return_value = {} cache_file = tmp_path / "cache.yaml" with patch("agent.model_metadata._get_context_cache_path", return_value=cache_file): save_context_length("custom/model", "http://local", 32768) # No base_url → cache skipped → falls to probe tier result = get_model_context_length("custom/model") assert result == CONTEXT_PROBE_TIERS[0] @patch("agent.model_metadata.fetch_model_metadata") @patch("agent.models_dev.lookup_models_dev_context", return_value=None) def test_stale_minimax_cache_32k_is_invalidated(self, mock_models_dev, mock_fetch, tmp_path): """Stale 32K cache entries for MiniMax must not keep tripping the 64K floor.""" mock_fetch.return_value = {} cache_file = tmp_path / "cache.yaml" base_url = "https://api.minimax.io/anthropic" with patch("agent.model_metadata._get_context_cache_path", return_value=cache_file): save_context_length("MiniMax-M2.7", base_url, 32768) result = get_model_context_length( "MiniMax-M2.7", base_url=base_url, provider="minimax", ) assert result == 204800 assert get_cached_context_length("MiniMax-M2.7", base_url) is None @patch("agent.models_dev.lookup_models_dev_context", return_value=None) @patch("agent.model_metadata.fetch_model_metadata") def test_openrouter_32k_underreport_for_minimax_falls_through_to_default(self, mock_fetch, mock_models_dev): """Unknown-provider fallback must reject stale OpenRouter 32K for MiniMax.""" mock_fetch.return_value = { "MiniMax-M2.7": {"context_length": 32768} } result = get_model_context_length("MiniMax-M2.7") assert result == 204800 @patch("agent.model_metadata.fetch_model_metadata") @patch("agent.models_dev.lookup_models_dev_context", return_value=None) def test_non_minimax_32k_cache_is_still_respected(self, mock_models_dev, mock_fetch, tmp_path): """The stale-32K invalidation must stay narrow and not touch unrelated models.""" mock_fetch.return_value = {} cache_file = tmp_path / "cache.yaml" base_url = "http://local" with patch("agent.model_metadata._get_context_cache_path", return_value=cache_file): save_context_length("qwen3.5:27b", base_url, 32768) result = get_model_context_length( "qwen3.5:27b", base_url=base_url, ) assert result == 32768 @patch("agent.model_metadata.fetch_model_metadata") @patch("agent.model_metadata.fetch_endpoint_model_metadata") def test_custom_endpoint_metadata_beats_fuzzy_default(self, mock_endpoint_fetch, mock_fetch): mock_fetch.return_value = {} mock_endpoint_fetch.return_value = { "zai-org/GLM-5-TEE": {"context_length": 65536} } result = get_model_context_length( "zai-org/GLM-5-TEE", base_url="https://llm.chutes.ai/v1", api_key="test-key", ) assert result == 65536 @patch("agent.model_metadata.fetch_model_metadata") @patch("agent.model_metadata.fetch_endpoint_model_metadata") def test_custom_endpoint_without_metadata_falls_back_to_catalog(self, mock_endpoint_fetch, mock_fetch): """Custom endpoint with no metadata should fall back to the hardcoded catalog (not 256K) when the model name matches a known entry. Previously this returned CONTEXT_PROBE_TIERS[0] (256K) because the custom-endpoint branch short-circuited before the catalog lookup. See #38865. """ mock_fetch.return_value = {} mock_endpoint_fetch.return_value = {} # GLM-5-TEE resolves through DEFAULT_CONTEXT_LENGTHS (longest matching GLM key), not the generic default. result = get_model_context_length( "zai-org/GLM-5-TEE", base_url="https://llm.chutes.ai/v1", api_key="test-key", ) from agent.model_metadata import DEFAULT_CONTEXT_LENGTHS, _longest_key_match assert result == _longest_key_match(DEFAULT_CONTEXT_LENGTHS, "zai-org/glm-5-tee")[1] # ── Local vs non-local Ollama context resolution (#63122) ────────── @patch("agent.model_metadata.get_cached_context_length", return_value=None) @patch("agent.model_metadata.fetch_model_metadata", return_value={}) @patch("agent.model_metadata._resolve_endpoint_context_length", return_value=None) @patch("agent.model_metadata._query_ollama_api_show", return_value=131072) @patch("agent.model_metadata._query_local_context_length", return_value=32768) @patch("agent.model_metadata.is_local_endpoint", return_value=True) @patch("agent.model_metadata.save_context_length") @patch("agent.model_metadata._maybe_cache_local_context_length") def test_local_ollama_prefers_num_ctx_over_gguf( self, mock_maybe_cache, mock_save, mock_is_local, mock_local_ctx, mock_ollama_show, mock_resolve_ep, mock_fetch, mock_cache, ): """Local Ollama: _query_local_context_length (num_ctx-first) must win over _query_ollama_api_show (GGUF-first). The configured Modelfile num_ctx is the context value the local probe prefers; the GGUF training max can be larger and would create a false-safe window for compression (#63122).""" result = get_model_context_length( "my-model", base_url="http://localhost:11434", ) assert result == 32768, ( f"Expected configured Modelfile num_ctx (32768), got {result}. " "Local Ollama must prefer num_ctx over GGUF training max." ) # The non-local-oriented probe must NOT fire when local probe succeeds mock_ollama_show.assert_not_called() # The local probe MUST be called exactly once mock_local_ctx.assert_called_once() # ── Codex routes are keyed on the transport, not the host (#116191) ────────────── @pytest.mark.parametrize( "provider, custom_providers", [ ("custom:codex-proxy", [{"name": "codex-proxy", "base_url": "http://127.0.0.1:8317/v1", "api_mode": "codex_responses"}]), ("openai-codex", None), # HERMES_CODEX_BASE_URL / model.base_url proxy per #115902 ], ) def test_codex_route_behind_proxy_resolves_codex_oauth_window(self, provider, custom_providers): """A Codex model served through a generic proxy URL must get the Codex OAuth window, not the direct-API catalog window: the compressor otherwise fires ~2x past the Codex limit.""" from agent import model_metadata as mm proxy_models = {"gpt-6-astra": {"id": "gpt-6-astra"}} # like CLIProxyAPI's /models: no context field with ( patch.object(mm, "get_cached_context_length", return_value=1_050_000), # stale pre-fix entry must not win patch.object(mm, "fetch_endpoint_model_metadata", return_value=proxy_models), patch.object(mm, "_query_ollama_api_show", return_value=None), patch.object(mm, "is_local_endpoint", return_value=False), patch.object(mm, "_fetch_codex_oauth_context_lengths_with_source", return_value=({}, False)), ): ctx = get_model_context_length( "gpt-6-astra", base_url="http://127.0.0.1:8317/v1", api_key="proxy-key", provider=provider, custom_providers=custom_providers, ) assert ctx == mm._CODEX_OAUTH_CONTEXT_FALLBACK["gpt-6-astra"] def test_codex_proxy_route_caps_an_opted_in_variant_by_its_own_catalog_max(self): """On a custom codex_responses route the -900k bump is capped by the gateway's own Codex-shaped max_context_window, asked once with the route's key at the route's URL; its context_window is not trusted, the base slug keeps the Codex table value, and a gateway without a Codex catalog is not re-probed on every resolution.""" from agent import model_metadata as mm custom = [{"name": "codex-proxy", "base_url": "http://127.0.0.1:8317/v1", "api_mode": "codex_responses"}] route = dict(base_url="http://127.0.0.1:8317/v1", api_key="proxy-key", provider="custom:codex-proxy", custom_providers=custom) codex_shaped = MagicMock(status_code=200) codex_shaped.json.return_value = {"models": [{"slug": "GPT-6-Sol", "context_window": 1_050_000, "max_context_window": 600_000}]} openai_shaped = MagicMock(status_code=200) openai_shaped.json.return_value = {"data": [{"id": "gpt-6-sol"}]} with patch.object(mm, "_codex_oauth_context_cache", {}), patch.object(mm, "_codex_oauth_max_context_cache", {}), \ patch.object(mm, "get_cached_context_length", return_value=None), patch.object(mm, "save_context_length"): with patch("agent.model_metadata.model_metadata_http.get", return_value=codex_shaped) as get: assert get_model_context_length("gpt-6-sol-900k", **route) == 600_000 assert get_model_context_length("gpt-6-sol", **route) == mm._CODEX_OAUTH_CONTEXT_FALLBACK["gpt-6-sol"] assert get.call_args_list and all(c.args[0].startswith("http://127.0.0.1:8317/v1/") for c in get.call_args_list) mm._codex_oauth_context_cache.clear() mm._codex_oauth_max_context_cache.clear() with patch("agent.model_metadata.model_metadata_http.get", return_value=openai_shaped) as get: for _ in range(3): assert get_model_context_length("gpt-6-sol-900k", **route) == 900_000 probes = get.call_count assert 0 < probes <= 2 # one probe (both client_version URLs at most), not one per resolution def test_codex_proxy_route_explicit_context_length_override_still_wins(self): """providers..models[].context_length beats the Codex table on a codex_responses route (#102644).""" custom = [{ "name": "codex-proxy", "base_url": "http://127.0.0.1:8317/v1", "api_mode": "codex_responses", "models": {"gpt-6-astra": {"context_length": 321_000}}, }] ctx = get_model_context_length( "gpt-6-astra", base_url="http://127.0.0.1:8317/v1", provider="custom:codex-proxy", custom_providers=custom, ) assert ctx == 321_000 # ========================================================================= # Bedrock context resolution — must run BEFORE custom-endpoint probe # ========================================================================= class TestBedrockContextResolution: """Regression tests for Bedrock context-length resolution order. Bug: because ``bedrock-runtime..amazonaws.com`` is not listed in ``_URL_TO_PROVIDER``, ``_is_known_provider_base_url`` returned False and the custom-endpoint probe at step 2 ran first — fetching ``/models`` from Bedrock (which it doesn't serve), returning the 128K default-fallback before execution ever reached the Bedrock branch. Fix: promote the Bedrock branch ahead of the custom-endpoint probe. """ @patch("agent.model_metadata.fetch_endpoint_model_metadata") def test_bedrock_claude_4_6_ignores_stale_200k_cache(self, mock_fetch, tmp_path): """Old 200K Bedrock cache entries must not mask the 1M table entry.""" cache_file = tmp_path / "context_length_cache.yaml" base_url = "https://bedrock-runtime.us-east-2.amazonaws.com" with patch("agent.model_metadata._get_context_cache_path", return_value=cache_file): save_context_length("us.anthropic.claude-sonnet-4-6", base_url, 200_000) ctx = get_model_context_length( "us.anthropic.claude-sonnet-4-6", provider="bedrock", base_url=base_url, ) assert ctx == 1_000_000 mock_fetch.assert_not_called() # ========================================================================= # Bedrock context cache persistence — only a probe result may be persisted # ========================================================================= class TestBedrockContextCachePersistence: """``_resolve_bedrock_context_length`` persisted whatever ``get_bedrock_context_length`` returned whenever a region was resolvable — and ``resolve_bedrock_region()`` always resolves one (it ends in ``or "us-east-1"``). A probe that returned None (expired SSO session, offline, opaque server error) therefore froze the static table value — the 128K default for a model with no table row — under ``model@`` or ``model@bedrock://`` (written as ``model@bedrock:``), and the probe, which the resolver treats as the only authoritative source, never ran for that model again. Invariants: only a probe-derived window may be persisted, and a failed probe is memoised in memory for ``_BEDROCK_PROBE_FAILURE_TTL_SECONDS`` so it is not re-sent on every resolution, yet runs again once the memo lapses. """ @pytest.fixture(autouse=True) def _clear_bedrock_probe_memo(self): """The failure memo is module-level state; it must not leak between tests.""" from agent import model_metadata as mm mm._BEDROCK_PROBE_FAILURE_CACHE.clear() yield mm._BEDROCK_PROBE_FAILURE_CACHE.clear() @patch("agent.bedrock_adapter.resolve_bedrock_region", return_value="us-east-1") @patch("agent.bedrock_adapter.probe_bedrock_context_length", return_value=None) def test_failed_probe_does_not_persist_static_fallback(self, mock_probe, mock_region, tmp_path): """A failed probe answers from the table (the 128K default here) but writes nothing to disk. On main this persists ``amazon.future-model-v1:0@bedrock:: 128000``.""" from agent.bedrock_adapter import BEDROCK_DEFAULT_CONTEXT_LENGTH model = "amazon.future-model-v1:0" # no BEDROCK_CONTEXT_LENGTHS row cache_file = tmp_path / "context_length_cache.yaml" with patch("agent.model_metadata._get_context_cache_path", return_value=cache_file): assert get_model_context_length(model, provider="bedrock") == BEDROCK_DEFAULT_CONTEXT_LENGTH assert get_cached_context_length(model, "bedrock://") is None assert not cache_file.exists() mock_probe.assert_called_once_with(model, "us-east-1") @patch("agent.bedrock_adapter.resolve_bedrock_region", return_value="us-east-1") @patch("agent.bedrock_adapter.probe_bedrock_context_length", side_effect=[None, 1_000_000]) def test_failed_probe_is_memoised_until_the_ttl_lapses(self, mock_probe, mock_region, tmp_path): """Not persisting must not turn the probe into a per-resolution cost: inside the TTL a second resolution answers from the table without re-probing and still writes nothing; once the memo lapses the probe runs again and its window is what gets persisted. On main the first call persists 128K and every later call serves it.""" from agent import model_metadata as mm from agent.bedrock_adapter import BEDROCK_DEFAULT_CONTEXT_LENGTH model = "amazon.future-model-v1:0" cache_file = tmp_path / "context_length_cache.yaml" with patch("agent.model_metadata._get_context_cache_path", return_value=cache_file): assert get_model_context_length(model, provider="bedrock") == BEDROCK_DEFAULT_CONTEXT_LENGTH assert get_model_context_length(model, provider="bedrock") == BEDROCK_DEFAULT_CONTEXT_LENGTH assert mock_probe.call_count == 1 assert not cache_file.exists() # memoised in memory only # Age the entry past the failure TTL (as tests/agent/test_probe_cache_followups.py does). for key in mm._BEDROCK_PROBE_FAILURE_CACHE: mm._BEDROCK_PROBE_FAILURE_CACHE[key] = ( time.monotonic() - mm._BEDROCK_PROBE_FAILURE_TTL_SECONDS - 1 ) assert get_model_context_length(model, provider="bedrock") == 1_000_000 assert get_cached_context_length(model, "bedrock://") == 1_000_000 assert mock_probe.call_count == 2 # ========================================================================= # _strip_provider_prefix — Ollama model:tag vs provider:model # ========================================================================= class TestStripProviderPrefix: def test_known_provider_prefix_is_stripped(self): assert _strip_provider_prefix("local:my-model") == "my-model" assert _strip_provider_prefix("openrouter:anthropic/claude-sonnet-4") == "anthropic/claude-sonnet-4" assert _strip_provider_prefix("anthropic:claude-sonnet-4") == "claude-sonnet-4" assert _strip_provider_prefix("stepfun:step-3.5-flash") == "step-3.5-flash" def test_http_urls_preserved(self): assert _strip_provider_prefix("http://example.com") == "http://example.com" assert _strip_provider_prefix("https://example.com") == "https://example.com" def test_registered_profile_name_and_alias_are_stripped(self, monkeypatch): import providers from providers.base import ProviderProfile monkeypatch.setattr(providers, "_REGISTRY", {}) monkeypatch.setattr(providers, "_ALIASES", {}) monkeypatch.setattr(providers, "_PROVIDER_LIST_CACHE", None) monkeypatch.setattr(providers, "_discovered", True) providers.register_provider( ProviderProfile(name="fake-provider", aliases=("fake-alias",)) ) assert _strip_provider_prefix("fake-provider:org/model") == "org/model" assert _strip_provider_prefix("fake-alias:org/model") == "org/model" def test_bundled_plugin_provider_prefix_is_stripped(self): assert _strip_provider_prefix("fireworks:accounts/fireworks/models/foo") == ( "accounts/fireworks/models/foo" ) def test_unknown_provider_prefix_is_unchanged(self): assert _strip_provider_prefix("not-a-provider:org/model") == ( "not-a-provider:org/model" ) def test_ollama_model_tag_is_unchanged(self): assert _strip_provider_prefix("qwen3.5:27b") == "qwen3.5:27b" @patch("agent.model_metadata.fetch_model_metadata") def test_ollama_model_tag_not_mangled_in_context_lookup(self, mock_fetch): """Ensure 'qwen3.5:27b' is NOT reduced to '27b' during context length lookup. We mock a custom endpoint that knows 'qwen3.5:27b' — the full name must reach the endpoint metadata lookup intact. """ mock_fetch.return_value = {} with patch("agent.model_metadata.fetch_endpoint_model_metadata") as mock_ep, \ patch("agent.model_metadata._is_custom_endpoint", return_value=True): mock_ep.return_value = {"qwen3.5:27b": {"context_length": 32768}} result = get_model_context_length( "qwen3.5:27b", base_url="http://localhost:11434/v1", ) assert result == 32768 # ========================================================================= # fetch_model_metadata — caching, TTL, slugs, failures # ========================================================================= class TestFetchModelMetadata: def _reset_cache(self): import agent.model_metadata as mm mm._model_metadata_cache = {} mm._model_metadata_cache_time = 0 def _isolate_disk_cache(self, monkeypatch, tmp_path): import agent.model_metadata as mm cache_path = tmp_path / "openrouter_model_metadata.json" monkeypatch.setattr(mm, "_get_model_metadata_cache_path", lambda: cache_path) return cache_path def test_network_success_writes_disk_cache(self, tmp_path, monkeypatch): self._reset_cache() cache_path = self._isolate_disk_cache(monkeypatch, tmp_path) mock_response = MagicMock() mock_response.json.return_value = { "data": [{"id": "live/model", "context_length": 67890, "name": "Live"}] } mock_response.raise_for_status = MagicMock() with patch("agent.model_metadata.model_metadata_http.get", return_value=mock_response): fetch_model_metadata(force_refresh=True) assert cache_path.exists() assert "live/model" in cache_path.read_text(encoding="utf-8") def test_network_failure_falls_back_to_stale_disk_cache(self, tmp_path, monkeypatch): self._reset_cache() cache_path = self._isolate_disk_cache(monkeypatch, tmp_path) cache_path.write_text( '{"stale/model":{"context_length":50000,"name":"Stale","pricing":{}}}', encoding="utf-8", ) old = time.time() - _MODEL_CACHE_TTL - 60 import os os.utime(cache_path, (old, old)) with patch("agent.model_metadata.model_metadata_http.get", side_effect=Exception("Network error")): result = fetch_model_metadata(force_refresh=True) assert result["stale/model"]["context_length"] == 50000 @patch("agent.model_metadata.model_metadata_http.get") def test_caches_result(self, mock_get): self._reset_cache() mock_response = MagicMock() mock_response.json.return_value = { "data": [{"id": "test/model", "context_length": 99999, "name": "Test"}] } mock_response.raise_for_status = MagicMock() mock_get.return_value = mock_response result1 = fetch_model_metadata(force_refresh=True) assert "test/model" in result1 assert mock_get.call_count == 1 result2 = fetch_model_metadata() assert "test/model" in result2 assert mock_get.call_count == 1 # cached @patch("agent.model_metadata.model_metadata_http.get") def test_canonical_slug_aliasing(self, mock_get): """Models with canonical_slug get indexed under both IDs.""" self._reset_cache() mock_response = MagicMock() mock_response.json.return_value = { "data": [{ "id": "anthropic/claude-3.5-sonnet:beta", "canonical_slug": "anthropic/claude-3.5-sonnet", "context_length": 200000, "name": "Claude 3.5 Sonnet" }] } mock_response.raise_for_status = MagicMock() mock_get.return_value = mock_response result = fetch_model_metadata(force_refresh=True) # Both the original ID and canonical slug should work assert "anthropic/claude-3.5-sonnet:beta" in result assert "anthropic/claude-3.5-sonnet" in result assert result["anthropic/claude-3.5-sonnet"]["context_length"] == 200000 # ========================================================================= # Context probe tiers # ========================================================================= class TestContextProbeTiers: def test_tiers_descending(self): for i in range(len(CONTEXT_PROBE_TIERS) - 1): assert CONTEXT_PROBE_TIERS[i] > CONTEXT_PROBE_TIERS[i + 1] class TestGetNextProbeTier: def test_from_256k(self): assert get_next_probe_tier(CONTEXT_PROBE_TIERS[0]) == CONTEXT_PROBE_TIERS[1] def test_from_8k_returns_none(self): assert get_next_probe_tier(CONTEXT_PROBE_TIERS[-1]) is None # ========================================================================= # Error message parsing # ========================================================================= class TestParseContextLimitFromError: @pytest.mark.parametrize("msg,expected", [ ("max_model_len 32768", 32768), ("max_model_len: 32768", 32768), ("max_model_len=32768", 32768), ("max_model_len (32768)", 32768), ("max_model_len is 32768", 32768), ("maximum model length 131072", 131072), ("maximum model length is 131072", 131072), ("maximum model length: 131072", 131072), ]) def test_vllm_delimiter_variants(self, msg, expected): """vLLM emits the limit with various delimiters (space/colon/equals/ paren/'is'). The parser must catch all of them — the original space-only patterns silently missed ':', '=', '(' and 'is' forms and fell through to None.""" assert parse_context_limit_from_error(msg) == expected @pytest.mark.parametrize("msg,expected", [ # Google Gemini/Gemma overflow phrasing (#57275): the limit follows # "supports up to"; the larger input count before it must NOT win. ("Unable to submit request because the input token count is 32825 " "but model only supports up to 32768. Reduce the input token count " "and try again.", 32768), ("input token count is 140000 but model only supports up to 131072", 131072), ("model supports up to 65536 tokens", 65536), ]) def test_google_supports_up_to_variants(self, msg, expected): """Google's overflow error was previously unparseable — recovery kept the wrong window and burned its attempts (#57275, residual claim 5).""" assert parse_context_limit_from_error(msg) == expected def test_google_supports_up_to_recalibrates_window(self): from agent.model_metadata import get_context_length_from_provider_error msg = ("Unable to submit request because the input token count is " "32825 but model only supports up to 32768.") assert get_context_length_from_provider_error(msg, 131072) == 32768 # Parsed limit not below current window → no recalibration. assert get_context_length_from_provider_error(msg, 32768) is None def test_output_cap_message_is_not_a_context_limit(self): """An output-cap error must never be cached as the context window (salvage #106769): the generic "limit ... of N" pattern matched Switchyard's message and clamped a >117K-context model to 16K on every later request.""" from agent.model_metadata import ( is_output_cap_error, parse_available_output_tokens_from_error, ) msg = "max_tokens cannot exceed the configured model output limit of 16384" assert parse_context_limit_from_error(msg) is None assert parse_available_output_tokens_from_error(msg) == 16384 assert is_output_cap_error(msg) # Genuine context messages still parse. assert parse_context_limit_from_error( "This model's maximum context length is 32768 tokens" ) == 32768 # ========================================================================= # Persistent context length cache # ========================================================================= class TestContextLengthCache: def test_non_positive_lengths_never_persisted(self, tmp_path): """save_context_length must refuse 0/negative values — a persisted 0 short-circuits step 1 (``0 is not None``) and poisons the whole resolution chain downstream (#25812).""" cache_file = tmp_path / "cache.yaml" with patch("agent.model_metadata._get_context_cache_path", return_value=cache_file): save_context_length("test/model", "http://x", 0) save_context_length("test/model", "http://x", -1) assert get_cached_context_length("test/model", "http://x") is None @patch("agent.model_metadata.fetch_model_metadata") def test_non_positive_cached_entry_dropped_and_reresolved(self, mock_fetch, tmp_path): """A pre-existing 0 entry (corrupted cache / manual edit) must be invalidated at step 1 and re-resolved instead of returned.""" mock_fetch.return_value = {} cache_file = tmp_path / "cache.yaml" with patch("agent.model_metadata._get_context_cache_path", return_value=cache_file): # Write the poison entry directly — save_context_length now refuses it. cache_file.write_text( "context_lengths:\n test/model@http://x: 0\n", encoding="utf-8" ) assert get_cached_context_length("test/model", "http://x") == 0 result = get_model_context_length("test/model", base_url="http://x") assert result > 0 assert get_cached_context_length("test/model", "http://x") != 0 def test_null_context_lengths_key_returns_empty(self, tmp_path): """``context_lengths:`` with no value parses as None — must behave like an empty cache instead of crashing every caller (#47135).""" cache_file = tmp_path / "cache.yaml" cache_file.write_text("context_lengths:\n", encoding="utf-8") with patch("agent.model_metadata._get_context_cache_path", return_value=cache_file): assert get_cached_context_length("test/model", "http://x") is None # save must also survive the null key and repair the file save_context_length("test/model", "http://x", 32768) assert get_cached_context_length("test/model", "http://x") == 32768 def test_idempotent_save(self, tmp_path): cache_file = tmp_path / "cache.yaml" with patch("agent.model_metadata._get_context_cache_path", return_value=cache_file): save_context_length("model", "http://x", 32768) save_context_length("model", "http://x", 32768) with open(cache_file, encoding="utf-8") as f: data = yaml.safe_load(f) assert len(data["context_lengths"]) == 1 @patch("agent.model_metadata.fetch_model_metadata") def test_cached_value_takes_priority(self, mock_fetch, tmp_path): mock_fetch.return_value = {} cache_file = tmp_path / "cache.yaml" with patch("agent.model_metadata._get_context_cache_path", return_value=cache_file): save_context_length("unknown/model", "http://local", 65536) assert get_model_context_length("unknown/model", base_url="http://local") == 65536 def test_write_failure_leaves_existing_cache_intact(self, tmp_path, monkeypatch): """An interrupted write must not corrupt or wipe the existing cache. The old non-atomic ``open(path, "w")`` truncated the file before dumping, so a crash/kill mid-write left empty or partial YAML — and the next load swallowed the error and returned ``{}``, silently wiping EVERY persisted context length. The atomic temp-file + ``os.replace`` write leaves the previous file byte-for-byte intact when the swap fails. """ import utils import agent.model_metadata as mm cache_file = tmp_path / "cache.yaml" monkeypatch.setattr(mm, "_get_context_cache_path", lambda: cache_file) # Seed a valid, populated cache. save_context_length("model-a", "http://a", 64000) original_bytes = cache_file.read_bytes() # Simulate a crash during the atomic swap step. def _boom(*_args, **_kwargs): raise OSError("simulated crash during atomic replace") monkeypatch.setattr(utils, "atomic_replace", _boom) # save_context_length is best-effort and swallows the error. save_context_length("model-b", "http://b", 128000) # Original file survives untouched — not truncated or emptied. assert cache_file.read_bytes() == original_bytes assert get_cached_context_length("model-a", "http://a") == 64000 # The failed write must not leave a stray temp file behind. assert list(cache_file.parent.glob(".cache_*.tmp")) == [] class TestGrok43StaleCacheGuard: """Pre-catalog builds resolved grok-4.3 via the generic 'grok-4' catch-all (256,000) and persisted it before the 'grok-4.3' (1M) catalog entry was added on 2026-05-15. The step-1 cache guard must drop that stale value and re-resolve to 1M, while leaving correct grok-4 entries (256,000) untouched. """ def test_stale_grok_4_3_dropped_and_reresolves_to_1m(self, tmp_path, monkeypatch): monkeypatch.setenv("HERMES_HOME", str(tmp_path)) import importlib import agent.model_metadata as mm importlib.reload(mm) base = "https://api.x.ai/v1" mm.save_context_length("grok-4.3", base, 256_000) ctx = mm.get_model_context_length( "grok-4.3", base_url=base, api_key="", provider="xai" ) assert ctx == 1_000_000 def test_grok_4_not_clobbered(self, tmp_path, monkeypatch): monkeypatch.setenv("HERMES_HOME", str(tmp_path)) import importlib import agent.model_metadata as mm importlib.reload(mm) base = "https://api.x.ai/v1" # 256,000 is the CORRECT value for plain grok-4 — guard must not touch it. for slug in ("grok-4", "grok-4-0709"): mm.save_context_length(slug, base, 256_000) ctx = mm.get_model_context_length( slug, base_url=base, api_key="", provider="xai" ) assert ctx == 256_000, f"{slug} should stay 256000, got {ctx}" class TestGenericPreCatalogStaleGuard: """Generic _stale_pre_catalog_cache_entry guard: models whose catalog entry postdates a shorter catch-all (qwen3.6-plus, grok-4-fast, grok-4.20, ...) get their pre-catalog cached values dropped, while correct or probe-derived values survive. Absorbs the per-model predicates and PR #37684's requested guards. """ def test_absorbed_pr_37684_models(self): from agent.model_metadata import _stale_pre_catalog_cache_entry # qwen3.6-plus (1M): old "qwen" catch-all persisted 131,072. assert _stale_pre_catalog_cache_entry("qwen3.6-plus", 131_072) assert _stale_pre_catalog_cache_entry("alibaba/qwen3.6-plus", 131_072) assert not _stale_pre_catalog_cache_entry("qwen3.6-plus", 1_048_576) # A 256K value for qwen3.6-plus is above the "qwen" catch-all — # could be a genuine probe result, so it is NOT dropped. assert not _stale_pre_catalog_cache_entry("qwen3.6-plus", 262_144) # Sibling qwen slugs with legitimately small windows are untouched. assert not _stale_pre_catalog_cache_entry("qwen3-coder", 131_072) def test_unknown_models_never_dropped(self): from agent.model_metadata import _stale_pre_catalog_cache_entry assert not _stale_pre_catalog_cache_entry("totally-unknown-model", 4096) assert not _stale_pre_catalog_cache_entry("minimax", 204_800) class TestMoAContextLength: """MoA virtual provider resolves context from the aggregator slot, not 256K default.""" def _write_moa_config( self, home, aggregator, custom_providers=None, providers=None ): import os os.makedirs(home, exist_ok=True) payload = { "moa": { "default_preset": "p", "presets": { "p": { "enabled": True, "reference_models": [ {"provider": "openrouter", "model": "openai/gpt-5.5"} ], "aggregator": aggregator, } }, } } if custom_providers is not None: payload["custom_providers"] = custom_providers if providers is not None: payload["providers"] = providers with open(os.path.join(home, "config.yaml"), "w", encoding="utf-8") as f: yaml.safe_dump(payload, f) def test_moa_resolves_from_aggregator(self, tmp_path, monkeypatch): home = str(tmp_path / ".hermes") monkeypatch.setenv("HERMES_HOME", home) self._write_moa_config(home, {"provider": "openrouter", "model": "anthropic/claude-opus-4.8"}) # The MoA preset name + virtual base_url would otherwise fall through to # the 256K default; instead it mirrors the aggregator's real window. agg_ctx = get_model_context_length( "anthropic/claude-opus-4.8", base_url="https://openrouter.ai/api/v1", provider="openrouter" ) moa_ctx = get_model_context_length("p", base_url="http://127.0.0.1/v1", provider="moa") assert moa_ctx == agg_ctx def test_moa_custom_context_configures_compressor_threshold( self, tmp_path, monkeypatch ): from agent.context_compressor import ContextCompressor configured_context = 600_000 home = str(tmp_path / ".hermes") monkeypatch.setenv("HERMES_HOME", home) self._write_moa_config( home, {"provider": "custom:example", "model": "example-model"}, providers={ "example": { "api": "http://127.0.0.1:1/v1", "default_model": "example-model", "models": { "example-model": { "context_length": configured_context, }, }, } }, ) with patch( "agent.model_metadata._resolve_endpoint_context_length", return_value=None, ) as endpoint_probe: compressor = ContextCompressor( model="p", base_url="http://127.0.0.1/v1", provider="moa", threshold_percent=0.50, quiet_mode=True, ) assert compressor.context_length == configured_context assert compressor.threshold_tokens == configured_context // 2 endpoint_probe.assert_not_called() # ========================================================================= # Fallback diagnostic logging # ========================================================================= class TestFallbackWarning: """When all 9 detection methods fail, the 10th fallback should log a warning so users with small-context models (8K, 32K) don't silently get 256K and hit hard-to-debug API context-length errors. The warning is deduped per (model, base_url) — the fallback result is deliberately never cached, so without dedup it would repeat on every resolution (e.g. once per gateway message via session hygiene). """ @pytest.fixture(autouse=True) def _reset_warned_set(self): from agent import model_metadata as mm mm._FALLBACK_WARNED.clear() yield mm._FALLBACK_WARNED.clear() @staticmethod def _patch_all_lookups(): from contextlib import ExitStack stack = ExitStack() for target, value in [ ("agent.model_metadata.get_cached_context_length", None), ("agent.model_metadata.fetch_model_metadata", {}), ("agent.model_metadata.fetch_endpoint_model_metadata", {}), ("agent.model_metadata._query_ollama_api_show", None), ("agent.model_metadata._query_anthropic_context_length", None), ("agent.model_metadata._endpoint_scoped_context_length", None), ("agent.model_metadata._resolve_endpoint_context_length", None), ("agent.models_dev.lookup_models_dev_context", None), ]: stack.enter_context(patch(target, return_value=value)) return stack def test_warning_emitted_on_fallback(self, caplog): import logging with self._patch_all_lookups(): with caplog.at_level(logging.WARNING, logger="agent.model_metadata"): result = get_model_context_length( "totally-unknown-model-xyz", ) assert result == DEFAULT_FALLBACK_CONTEXT # The warning must mention the model name and the config override hint. warning_msgs = [r for r in caplog.records if r.levelno == logging.WARNING] assert any("totally-unknown-model-xyz" in r.getMessage() for r in warning_msgs) assert any("model.context_length" in r.getMessage() for r in warning_msgs) def test_warning_fires_once_per_model(self, caplog): """Repeated resolutions of the same unknown model warn only once.""" import logging with self._patch_all_lookups(): with caplog.at_level(logging.WARNING, logger="agent.model_metadata"): for _ in range(3): get_model_context_length("totally-unknown-model-xyz") fallback_warnings = [ r for r in caplog.records if r.levelno == logging.WARNING and "falling back" in r.getMessage() ] assert len(fallback_warnings) == 1 def test_warning_emitted_on_custom_endpoint_fallback(self, caplog): """The sibling step-3b fallback (custom/local endpoint, probes down, no catalog match) is the same silent-256K bug class and must warn too.""" import logging with self._patch_all_lookups(), \ patch("agent.model_metadata._query_local_context_length", return_value=None): with caplog.at_level(logging.WARNING, logger="agent.model_metadata"): result = get_model_context_length( "totally-unknown-model-xyz", base_url="http://192.168.1.50:8080/v1", ) assert result == DEFAULT_FALLBACK_CONTEXT warning_msgs = [r for r in caplog.records if r.levelno == logging.WARNING] assert any("totally-unknown-model-xyz" in r.getMessage() for r in warning_msgs) assert any("model.context_length" in r.getMessage() for r in warning_msgs) def test_no_warning_when_cached(self, caplog): """No fallback warning when the context length is found in the cache.""" import logging with patch( "agent.model_metadata.get_cached_context_length", return_value=32_000, ): with caplog.at_level(logging.WARNING, logger="agent.model_metadata"): result = get_model_context_length( "some-model", base_url="http://127.0.0.1:1/v1", ) assert result == 32_000 fallback_warnings = [ r for r in caplog.records if r.levelno == logging.WARNING and "falling back" in r.getMessage() ] assert len(fallback_warnings) == 0 # ========================================================================= # get_model_context_length — OpenRouter routing-variant suffixes # ========================================================================= class TestOpenRouterRoutingVariantContextLength: """`:nitro`/`:floor`/`:exacto`/`:online` are request-time routing modifiers, not catalog models: /models lists only the base id and the variant runs the same model, so a variant must resolve to whatever its base resolves to instead of a generic family default (#97820). `:free`/`:batch` are real SKUs with their own windows and must NOT be stripped.""" _CATALOG = { "x-ai/grok-4.6": {"context_length": 2_000_000}, "thinkingmachines/inkling": {"context_length": 1_000_000}, "thinkingmachines/inkling:free": {"context_length": 64_000}, } @pytest.mark.parametrize("suffix", ["nitro", "floor", "exacto", "online"]) @patch("agent.model_metadata.get_cached_context_length", return_value=None) @patch("agent.models_dev.lookup_models_dev_context", return_value=None) @patch("agent.model_metadata.fetch_model_metadata") def test_variant_matches_base_but_real_sku_keeps_own_window( self, mock_fetch, mock_models_dev, mock_cache, suffix ): mock_fetch.return_value = self._CATALOG base_ctx = get_model_context_length("x-ai/grok-4.6", provider="openrouter") variant_ctx = get_model_context_length(f"x-ai/grok-4.6:{suffix}", provider="openrouter") assert variant_ctx == base_ctx == 2_000_000 assert variant_ctx != DEFAULT_CONTEXT_LENGTHS.get("grok") assert get_model_context_length("thinkingmachines/inkling:free", provider="openrouter") == 64_000 def test_endpoint_pricing_already_per_million_is_not_inflated(): """#112018 / #34256 / #79174: a /models catalog quoting USD per 1M tokens (with or without an explicit ``unit``) must reach usage_pricing as per-token rates, so the cost estimate is $0.60/M — not $600,000/M.""" from agent import model_metadata as mm from agent import usage_pricing as up catalog = { "minimax-m3": {"id": "minimax-m3", "pricing": {"currency": "USD", "prompt": 0.6, "completion": 1.2, "cache_read": 0.12}}, "glm-x": {"id": "glm-x", "pricing": {"currency": "CNY", "unit": "per_1m_tokens", "prompt": 1, "completion": 2}}, "crof-a": {"id": "crof-a", "cost": {"input": 0.04, "output": 0.15}}, } meta = {mid: mm._endpoint_model_entry(model, mid, None) for mid, model in catalog.items()} dollars_per_million = { mid: up._pricing_entry_from_metadata(meta, mid, source_url="x", pricing_version="openai-compatible-models-api") for mid in catalog } assert float(dollars_per_million["minimax-m3"].input_cost_per_million) == pytest.approx(0.6) assert float(dollars_per_million["minimax-m3"].cache_read_cost_per_million) == pytest.approx(0.12) assert float(dollars_per_million["glm-x"].output_cost_per_million) == pytest.approx(2.0) assert float(dollars_per_million["crof-a"].input_cost_per_million) == pytest.approx(0.04) def test_endpoint_pricing_per_token_quotes_pass_through_unchanged(): """Control: per-token quotes (the OpenRouter convention) and per-request fees are left alone.""" from agent import model_metadata as mm from agent import usage_pricing as up model = {"id": "m", "pricing": {"prompt": "0.0000006", "completion": "0.0000012", "request": "0.005"}} meta = {"m": mm._endpoint_model_entry(model, "m", None)} entry = up._pricing_entry_from_metadata(meta, "m", source_url="x", pricing_version="openai-compatible-models-api") assert float(entry.input_cost_per_million) == pytest.approx(0.6) assert float(entry.output_cost_per_million) == pytest.approx(1.2) assert float(entry.request_cost) == pytest.approx(0.005)