refactor(gateway): phase helpers for _resolve_gateway_model_context (differentially fuzzed, 3072 cases)

This commit is contained in:
Teknium
2026-09-02 19:08:44 -07:00
parent 9610173c22
commit b64d0c7602
+38 -51
View File
@@ -2516,68 +2516,55 @@ def _resolve_gateway_model_context(model: Optional[str] = None) -> _GatewayModel
from agent.model_metadata import DEFAULT_FALLBACK_CONTEXT, get_model_context_length
resolved_model = model or _resolve_gateway_model()
config_context_length = None
provider = None
base_url = None
api_key = None
custom_providers = None
configured_model = None
configured_provider = None
configured_base_url = None
config_context_length = provider = base_url = api_key = custom_providers = None
configured_model = configured_provider = configured_base_url = None
try:
def _read_config() -> None:
nonlocal config_context_length, provider, base_url, custom_providers
nonlocal configured_model, configured_provider, configured_base_url
data = _load_gateway_config()
if data:
model_cfg = data.get("model", {})
if isinstance(model_cfg, dict):
configured_model = model_cfg.get("default") or model_cfg.get("model")
raw_ctx = model_cfg.get("context_length")
if raw_ctx is not None:
with suppress(TypeError, ValueError):
config_context_length = int(raw_ctx)
provider = model_cfg.get("provider") or None
base_url = model_cfg.get("base_url") or None
configured_provider = provider
configured_base_url = base_url
try:
from hermes_cli.config import get_compatible_custom_providers
if not data:
return
model_cfg = data.get("model", {})
if isinstance(model_cfg, dict):
configured_model = model_cfg.get("default") or model_cfg.get("model")
raw_ctx = model_cfg.get("context_length")
if raw_ctx is not None:
with suppress(TypeError, ValueError):
config_context_length = int(raw_ctx)
configured_provider = provider = model_cfg.get("provider") or None
configured_base_url = base_url = model_cfg.get("base_url") or None
try:
from hermes_cli.config import get_compatible_custom_providers
custom_providers = get_compatible_custom_providers(data)
except Exception:
custom_providers = data.get("custom_providers")
custom_providers = get_compatible_custom_providers(data)
except Exception:
custom_providers = data.get("custom_providers")
except Exception:
pass
try:
def _read_runtime() -> None:
nonlocal provider, base_url, api_key
runtime = _resolve_runtime_agent_kwargs()
provider = runtime.get("provider") or provider
base_url = runtime.get("base_url") or base_url
api_key = runtime.get("api_key")
except Exception:
pass
if config_context_length is not None:
try:
from hermes_cli.route_identity import should_clear_context_pin
def _pin_still_applies() -> bool:
# A configured context_length pin is dropped when the effective route no longer matches the
# configured one; an unresolvable check also drops it.
from hermes_cli.route_identity import should_clear_context_pin
return not should_clear_context_pin(
configured_model, resolved_model, configured_base_url, base_url, configured_provider, provider)
if should_clear_context_pin(
configured_model, resolved_model, configured_base_url, base_url,
configured_provider, provider):
config_context_length = None
except Exception:
config_context_length = None
def _custom_ctx() -> Optional[int]:
from hermes_cli.config import get_custom_provider_context_length
return get_custom_provider_context_length(
model=resolved_model, base_url=base_url, custom_providers=custom_providers)
_best_effort(_read_config)
_best_effort(_read_runtime)
if config_context_length is not None and not _best_effort(_pin_still_applies):
config_context_length = None
if config_context_length is None and custom_providers and base_url:
try:
from hermes_cli.config import get_custom_provider_context_length
custom_ctx = get_custom_provider_context_length(
model=resolved_model, base_url=base_url, custom_providers=custom_providers
)
if custom_ctx:
config_context_length = custom_ctx
except Exception:
pass
config_context_length = _best_effort(_custom_ctx) or None
context_length = get_model_context_length(
resolved_model, base_url=base_url or "", api_key=api_key or "",