from __future__ import annotations import pytest from opensquilla.gateway.config import GatewayConfig from opensquilla.provider.ensemble import build_ensemble_provider_from_config from opensquilla.provider.selector import ProviderConfig def test_llm_ensemble_defaults_to_disabled_for_model_router_first_install() -> None: cfg = GatewayConfig() ensemble = cfg.llm_ensemble assert cfg.squilla_router.enabled is True assert ensemble.enabled is False assert ensemble.mode == "b5_fusion" assert ensemble.selection_mode == "static_openrouter_b5" assert ensemble.proposer_tools is False assert ensemble.min_successful_proposers == 1 assert ensemble.model_options == [] assert ensemble.candidates == [] assert ensemble.candidate_max_chars == 24_000 assert ensemble.proposer_timeout_seconds == 3600.0 assert ensemble.aggregator_timeout_seconds == 3600.0 assert ensemble.shuffle_candidates is True assert ensemble.record_candidates is False enabled_cfg = cfg.model_copy(deep=True) enabled_cfg.llm_ensemble.enabled = True provider = build_ensemble_provider_from_config( config=enabled_cfg, inherited_provider_config=ProviderConfig( provider="openrouter", model="routed/model", api_key="fake", base_url="https://openrouter.example/api/v1", ), fallback_provider=None, turn_metadata={"routed_tier": "c0"}, ) assert provider.profile_name == "static_openrouter_b5" assert [member.provider_config.model for member in provider.proposers] == [ "deepseek/deepseek-v4-pro", "z-ai/glm-5.2", "moonshotai/kimi-k2.7-code", "qwen/qwen3.7-max", ] assert provider.aggregator.provider_config.model == "z-ai/glm-5.2" assert provider.min_successful_proposers == 3 assert provider.proposer_timeout_seconds == 300.0 assert provider.aggregator_timeout_seconds == 480.0 assert provider.shuffle_candidates is False assert provider.quorum_grace_seconds == 30.0 def test_static_openrouter_b5_does_not_need_model_options() -> None: cfg = GatewayConfig( llm_ensemble={ "enabled": True, "selection_mode": "static_openrouter_b5", "model_options": [], } ) provider = build_ensemble_provider_from_config( config=cfg, inherited_provider_config=ProviderConfig( provider="openrouter", model="routed/model", api_key="fake", base_url="https://openrouter.example/api/v1", ), fallback_provider=None, ) assert provider.profile_name == "static_openrouter_b5" assert [member.provider_config.model for member in provider.proposers] == [ "deepseek/deepseek-v4-pro", "z-ai/glm-5.2", "moonshotai/kimi-k2.7-code", "qwen/qwen3.7-max", ] assert provider.aggregator.provider_config.model == "z-ai/glm-5.2" def test_static_tokenrhythm_b5_mirrors_the_openrouter_lineup() -> None: cfg = GatewayConfig( llm_ensemble={ "enabled": True, "selection_mode": "static_tokenrhythm_b5", } ) provider = build_ensemble_provider_from_config( config=cfg, inherited_provider_config=ProviderConfig( provider="tokenrhythm", model="deepseek-v4-pro", api_key="fake", base_url="https://tokenrhythm.example/v1", ), fallback_provider=None, ) assert provider.profile_name == "static_tokenrhythm_b5" assert [member.provider_config.model for member in provider.proposers] == [ "deepseek-v4-pro", "glm-5.2", "kimi-k2.7-code", "qwen3.7-max", ] assert all( member.provider_config.provider == "tokenrhythm" for member in provider.proposers ) assert provider.aggregator.provider_config.provider == "tokenrhythm" assert provider.aggregator.provider_config.model == "glm-5.2" # Same aggregation defaults as the static OpenRouter profile. assert provider.min_successful_proposers == 3 assert provider.proposer_timeout_seconds == 300.0 assert provider.aggregator_timeout_seconds == 480.0 assert provider.shuffle_candidates is False assert provider.quorum_grace_seconds == 30.0 def test_static_b5_mode_tables_agree_across_gateway_and_provider() -> None: # gateway must not be imported from provider, so the selection-mode → # provider table exists on both sides; this pins them together. from typing import get_args from opensquilla.gateway.config import ( STATIC_B5_SELECTION_MODE_PROVIDERS, LlmEnsembleConfig, ) from opensquilla.provider.ensemble import STATIC_B5_PROFILES assert { mode: profile.provider_id for mode, profile in STATIC_B5_PROFILES.items() } == STATIC_B5_SELECTION_MODE_PROVIDERS literal_modes = set( get_args(LlmEnsembleConfig.model_fields["selection_mode"].annotation) ) assert literal_modes == { "router_dynamic", "custom_b5", *STATIC_B5_SELECTION_MODE_PROVIDERS, } def test_router_dynamic_ensemble_allows_empty_custom_model_options() -> None: cfg = GatewayConfig( llm_ensemble={ "selection_mode": "router_dynamic", "model_options": [], } ) assert cfg.llm_ensemble.model_options == [] def test_router_dynamic_ignores_legacy_default_openrouter_model_options() -> None: cfg = GatewayConfig( llm={ "provider": "deepseek", "model": "deepseek-v4-flash", "api_key": "fake", "base_url": "https://api.deepseek.com", }, llm_ensemble={ "enabled": True, "selection_mode": "router_dynamic", "model_options": [ "deepseek/deepseek-v4-pro", "z-ai/glm-5.2", "qwen/qwen3.7-plus", "deepseek/deepseek-v4-flash", "qwen/qwen3.7-max", "moonshotai/kimi-k2.6", "moonshotai/kimi-k2.7-code", "minimax/minimax-m3", ], }, squilla_router={ "enabled": True, "tiers": { "c0": {"provider": "deepseek", "model": "deepseek-v4-flash"}, "c1": {"provider": "deepseek", "model": "deepseek-v4-flash"}, "c2": {"provider": "deepseek", "model": "deepseek-v4-pro"}, "c3": {"provider": "deepseek", "model": "deepseek-v4-pro"}, }, }, ) inherited = ProviderConfig( provider="deepseek", model="deepseek-v4-flash", api_key="fake", base_url="https://api.deepseek.com", ) provider = build_ensemble_provider_from_config( config=cfg, inherited_provider_config=inherited, fallback_provider=None, turn_metadata={"routed_tier": "c1"}, ) pool = provider.selection_plan["candidate_pool"] assert all(candidate["source"] != "legacy_model_options" for candidate in pool) assert all(candidate["provider"] != "openrouter" for candidate in pool) def test_llm_ensemble_validates_selection_mode() -> None: with pytest.raises(ValueError, match="selection_mode"): GatewayConfig(llm_ensemble={"selection_mode": "static_unknown"}) def test_llm_ensemble_model_options_are_operator_configurable() -> None: cfg = GatewayConfig( llm_ensemble={ "model_options": [" custom/model ", "custom/model", "other/model"], } ) assert cfg.llm_ensemble.model_options == ["custom/model", "other/model"] def test_router_dynamic_keeps_non_default_legacy_model_options_with_source() -> None: cfg = GatewayConfig( llm_ensemble={ "enabled": True, "selection_mode": "router_dynamic", "model_options": ["vendor/custom-model"], } ) inherited = ProviderConfig( provider="deepseek", model="deepseek-v4-flash", api_key="fake", base_url="https://api.deepseek.com", ) provider = build_ensemble_provider_from_config( config=cfg, inherited_provider_config=inherited, fallback_provider=None, turn_metadata={"routed_tier": "c1"}, ) pool = provider.selection_plan["candidate_pool"] legacy = next(candidate for candidate in pool if candidate["model"] == "vendor/custom-model") assert legacy["provider"] == "openrouter" assert legacy["source"] == "legacy_model_options" def test_router_dynamic_uses_structured_candidates_with_source() -> None: cfg = GatewayConfig( llm_ensemble={ "enabled": True, "selection_mode": "router_dynamic", "candidates": [ { "provider": "openrouter", "model": "qwen/qwen3.7-max", "source": "custom", "enabled": True, }, { "provider": "openrouter", "model": "disabled/model", "source": "custom", "enabled": False, }, ], } ) inherited = ProviderConfig( provider="deepseek", model="deepseek-v4-flash", api_key="fake", base_url="https://api.deepseek.com", ) provider = build_ensemble_provider_from_config( config=cfg, inherited_provider_config=inherited, fallback_provider=None, turn_metadata={"routed_tier": "c2"}, ) pool = provider.selection_plan["candidate_pool"] assert any( candidate["provider"] == "openrouter" and candidate["model"] == "qwen/qwen3.7-max" and candidate["source"] == "custom" for candidate in pool ) assert all(candidate["model"] != "disabled/model" for candidate in pool) def test_build_ensemble_provider_inherits_current_openrouter_credentials() -> None: cfg = GatewayConfig(llm_ensemble={"enabled": True}) inherited = ProviderConfig( provider="openrouter", model="routed/model", api_key="fake", base_url="https://openrouter.example/api/v1", proxy="http://proxy.local:7890", provider_routing={"z-ai/glm-5.2": "z-ai"}, ) provider = build_ensemble_provider_from_config( config=cfg, inherited_provider_config=inherited, fallback_provider=None, ) members = [*provider.proposers, provider.aggregator] assert all(member.provider_config.api_key == "fake" for member in members) assert all( member.provider_config.base_url == "https://openrouter.example/api/v1" for member in members ) assert all(member.provider_config.proxy == "http://proxy.local:7890" for member in members) assert provider.aggregator.provider_config.provider_routing == {"z-ai/glm-5.2": "z-ai"} def test_router_dynamic_ensemble_uses_small_c0_slot_template() -> None: cfg = GatewayConfig( llm_ensemble={ "enabled": True, "selection_mode": "router_dynamic", "min_successful_proposers": 4, } ) inherited = ProviderConfig( provider="openrouter", model="deepseek/deepseek-v4-flash", api_key="fake", base_url="https://openrouter.example/api/v1", ) provider = build_ensemble_provider_from_config( config=cfg, inherited_provider_config=inherited, fallback_provider=None, turn_metadata={"routed_tier": "c0", "routing_confidence": 0.93}, ) assert provider.profile_name == "router_dynamic/c0" assert [member.label for member in provider.proposers] == ["anchor", "cheap_contrast"] assert [member.provider_config.model for member in provider.proposers][0] == ( "deepseek/deepseek-v4-flash" ) assert len(provider.proposers) == 2 assert provider.min_successful_proposers == 2 assert provider.selection_plan["slot_template"] == ["anchor", "cheap_contrast"] assert provider.selection_plan["aggregator_slot"] == "aggregator_fast" assert provider.selection_plan["duplicate_policy"] == "selected_penalty" assert provider.proposer_timeout_seconds == 3600.0 assert provider.aggregator_timeout_seconds == 3600.0 assert provider.quorum_grace_seconds == 0.0 def test_router_dynamic_ensemble_uses_slot_specific_c2_selection() -> None: cfg = GatewayConfig( llm_ensemble={ "enabled": True, "selection_mode": "router_dynamic", "model_options": [ "deepseek/deepseek-v4-pro", "z-ai/glm-5.2", "google/gemini-3-flash-preview", "qwen/qwen3.7-plus", "anthropic/claude-opus-4.8", ], } ) inherited = ProviderConfig( provider="openrouter", model="z-ai/glm-5.2", api_key="fake", base_url="https://openrouter.example/api/v1", ) provider = build_ensemble_provider_from_config( config=cfg, inherited_provider_config=inherited, fallback_provider=None, turn_metadata={"routed_tier": "c2", "routing_confidence": 0.82}, ) assert provider.profile_name == "router_dynamic/c2" assert [member.label for member in provider.proposers] == [ "anchor", "adjacent_tier_check", "orthogonal_family", ] assert provider.proposers[0].provider_config.model == "z-ai/glm-5.2" assert provider.selection_plan["aggregator_slot"] == "aggregator_strong" assert provider.selection_plan["slots"][1]["slot"] == "adjacent_tier_check" assert provider.selection_plan["slots"][2]["slot"] == "orthogonal_family" assert provider.selection_plan["aggregator"]["slot"] == "aggregator_strong" assert provider.selection_plan["candidate_pool_size"] >= 5 def test_static_openrouter_b5_ensemble_locks_members_across_routed_tiers() -> None: cfg = GatewayConfig( llm_ensemble={ "enabled": True, "selection_mode": "static_openrouter_b5", "min_successful_proposers": 9, "shuffle_candidates": False, } ) inherited = ProviderConfig( provider="openrouter", model="routed/model", api_key="fake", base_url="https://openrouter.example/api/v1", proxy="http://proxy.local:7890", provider_routing={"z-ai/glm-5.2": "z-ai"}, ) expected_proposers = [ "deepseek/deepseek-v4-pro", "z-ai/glm-5.2", "moonshotai/kimi-k2.7-code", "qwen/qwen3.7-max", ] for tier in ("c0", "c1", "c2", "c3"): provider = build_ensemble_provider_from_config( config=cfg, inherited_provider_config=inherited, fallback_provider=None, turn_metadata={"routed_tier": tier, "routing_confidence": 0.99}, ) assert provider.profile_name == "static_openrouter_b5" assert [member.provider_config.model for member in provider.proposers] == expected_proposers assert provider.aggregator.provider_config.model == "z-ai/glm-5.2" assert provider.selection_plan == { "strategy": "static_openrouter_b5", "profile": "static_openrouter_b5", "proposer_models": expected_proposers, "aggregator_model": "z-ai/glm-5.2", "proposer_count": 4, "configured_min_successful_proposers": 9, "effective_min_successful_proposers": 4, "configured_proposer_timeout_seconds": 3600.0, "effective_proposer_timeout_seconds": 300.0, "configured_aggregator_timeout_seconds": 3600.0, "effective_aggregator_timeout_seconds": 480.0, "configured_shuffle_candidates": False, "effective_shuffle_candidates": False, "quorum_grace_seconds": 30.0, } assert provider.min_successful_proposers == 4 assert provider.proposer_timeout_seconds == 300.0 assert provider.aggregator_timeout_seconds == 480.0 assert provider.shuffle_candidates is False assert provider.quorum_grace_seconds == 30.0 members = [*provider.proposers, provider.aggregator] assert all(member.provider_config.provider == "openrouter" for member in members) assert all(member.provider_config.api_key == "fake" for member in members) assert all( member.provider_config.base_url == "https://openrouter.example/api/v1" for member in members ) assert all(member.provider_config.proxy == "http://proxy.local:7890" for member in members) assert all( member.provider_config.provider_routing == {"z-ai/glm-5.2": "z-ai"} for member in members ) def test_static_openrouter_b5_ensemble_uses_profile_effective_defaults() -> None: cfg = GatewayConfig( llm_ensemble={ "enabled": True, "selection_mode": "static_openrouter_b5", } ) provider = build_ensemble_provider_from_config( config=cfg, inherited_provider_config=ProviderConfig( provider="openrouter", model="routed/model", api_key="fake", base_url="https://openrouter.example/api/v1", ), fallback_provider=None, ) assert cfg.llm_ensemble.min_successful_proposers == 1 assert cfg.llm_ensemble.proposer_timeout_seconds == 3600.0 assert cfg.llm_ensemble.aggregator_timeout_seconds == 3600.0 assert cfg.llm_ensemble.shuffle_candidates is True assert provider.min_successful_proposers == 3 assert provider.proposer_timeout_seconds == 300.0 assert provider.aggregator_timeout_seconds == 480.0 assert provider.shuffle_candidates is False assert provider.quorum_grace_seconds == 30.0 def test_static_openrouter_b5_ensemble_preserves_custom_effective_values() -> None: cfg = GatewayConfig( llm_ensemble={ "enabled": True, "selection_mode": "static_openrouter_b5", "min_successful_proposers": 2, "proposer_timeout_seconds": 180.0, "aggregator_timeout_seconds": 900.0, "shuffle_candidates": False, } ) provider = build_ensemble_provider_from_config( config=cfg, inherited_provider_config=ProviderConfig( provider="openrouter", model="routed/model", api_key="fake", base_url="https://openrouter.example/api/v1", ), fallback_provider=None, ) assert provider.min_successful_proposers == 2 assert provider.proposer_timeout_seconds == 180.0 assert provider.aggregator_timeout_seconds == 900.0 assert provider.shuffle_candidates is False def _custom_b5_config(**overrides: object) -> GatewayConfig: payload: dict[str, object] = { "enabled": True, "selection_mode": "custom_b5", "candidates": [ {"provider": "volcengine", "model": "doubao-2.0-pro", "role": "primary"}, {"provider": "volcengine", "model": "deepseek-v4-flash", "role": "fast_check"}, {"provider": "volcengine", "model": "kimi-k2.6", "role": "contrast"}, {"provider": "volcengine", "model": "deepseek-v4-pro", "role": "aggregator"}, ], } payload.update(overrides) return GatewayConfig(llm_ensemble=payload) def _volcengine_inherited() -> ProviderConfig: return ProviderConfig( provider="volcengine", model="deepseek-v4-pro", api_key="fake", base_url="https://volcengine.example/api/v3", ) def test_custom_b5_builds_role_labelled_proposers_and_single_aggregator() -> None: provider = build_ensemble_provider_from_config( config=_custom_b5_config(), inherited_provider_config=_volcengine_inherited(), fallback_provider=None, ) assert provider.profile_name == "custom_b5" assert [member.label for member in provider.proposers] == [ "primary", "fast_check", "contrast", ] assert [member.provider_config.model for member in provider.proposers] == [ "doubao-2.0-pro", "deepseek-v4-flash", "kimi-k2.6", ] assert provider.aggregator.provider_config.model == "deepseek-v4-pro" assert provider.selection_plan["aggregator"]["source"] == "candidate_role" def test_custom_b5_uses_fixed_lineup_effective_defaults_with_auto_quorum() -> None: cfg = _custom_b5_config() provider = build_ensemble_provider_from_config( config=cfg, inherited_provider_config=_volcengine_inherited(), fallback_provider=None, ) # Stored legacy defaults are replaced by the fixed-lineup family; quorum # is derived as N-1 for the 3-proposer lineup. assert cfg.llm_ensemble.min_successful_proposers == 1 assert provider.min_successful_proposers == 2 assert provider.proposer_timeout_seconds == 300.0 assert provider.aggregator_timeout_seconds == 480.0 assert provider.shuffle_candidates is False assert provider.quorum_grace_seconds == 30.0 def test_custom_b5_preserves_explicit_quorum_and_timeouts() -> None: cfg = _custom_b5_config( min_successful_proposers=3, proposer_timeout_seconds=120.0, aggregator_timeout_seconds=600.0, ) provider = build_ensemble_provider_from_config( config=cfg, inherited_provider_config=_volcengine_inherited(), fallback_provider=None, ) assert provider.min_successful_proposers == 3 assert provider.proposer_timeout_seconds == 120.0 assert provider.aggregator_timeout_seconds == 600.0 def test_custom_b5_without_aggregator_row_inherits_the_routed_model() -> None: cfg = GatewayConfig( llm_ensemble={ "enabled": True, "selection_mode": "custom_b5", "candidates": [ {"provider": "volcengine", "model": "doubao-2.0-pro"}, {"provider": "volcengine", "model": "kimi-k2.6"}, ], } ) provider = build_ensemble_provider_from_config( config=cfg, inherited_provider_config=_volcengine_inherited(), fallback_provider=None, ) assert provider.aggregator.provider_config.model == "deepseek-v4-pro" assert provider.selection_plan["aggregator"]["source"] == "inherited_model" def test_custom_b5_disabled_candidates_are_excluded_from_the_lineup() -> None: cfg = GatewayConfig( llm_ensemble={ "enabled": True, "selection_mode": "custom_b5", "candidates": [ {"provider": "volcengine", "model": "doubao-2.0-pro"}, {"provider": "volcengine", "model": "kimi-k2.6"}, {"provider": "volcengine", "model": "deepseek-v4-flash", "enabled": False}, ], } ) provider = build_ensemble_provider_from_config( config=cfg, inherited_provider_config=_volcengine_inherited(), fallback_provider=None, ) assert [member.provider_config.model for member in provider.proposers] == [ "doubao-2.0-pro", "kimi-k2.6", ] def test_custom_b5_validation_rejects_undersized_and_oversized_lineups() -> None: with pytest.raises(Exception, match="at least 2"): GatewayConfig( llm_ensemble={ "enabled": True, "selection_mode": "custom_b5", "candidates": [{"provider": "a", "model": "m1"}], } ) with pytest.raises(Exception, match="at most 6"): GatewayConfig( llm_ensemble={ "enabled": True, "selection_mode": "custom_b5", "candidates": [ {"provider": "a", "model": f"m{i}"} for i in range(7) ], } ) def test_custom_b5_validation_rejects_quorum_above_proposer_count() -> None: with pytest.raises(Exception, match="min_successful_proposers"): GatewayConfig( llm_ensemble={ "enabled": True, "selection_mode": "custom_b5", "min_successful_proposers": 4, "candidates": [ {"provider": "a", "model": "m1"}, {"provider": "a", "model": "m2"}, ], } ) def test_candidate_roles_normalize_and_reject_dual_aggregators() -> None: cfg = GatewayConfig( llm_ensemble={ "candidates": [ {"provider": "a", "model": "m1", "role": "AGGREGATOR"}, {"provider": "a", "model": "m2", "role": "definitely-not-a-role"}, ], } ) assert cfg.llm_ensemble.candidates[0].role == "aggregator" # Unknown roles coerce to unassigned instead of failing gateway boot. assert cfg.llm_ensemble.candidates[1].role == "" with pytest.raises(Exception, match="at most one"): GatewayConfig( llm_ensemble={ "candidates": [ {"provider": "a", "model": "m1", "role": "aggregator"}, {"provider": "a", "model": "m2", "role": "aggregator"}, ], } ) def test_custom_b5_lineup_ready_gates_on_member_credentials( monkeypatch: pytest.MonkeyPatch, ) -> None: from opensquilla.provider.ensemble import custom_b5_lineup_ready monkeypatch.delenv("OPENROUTER_API_KEY", raising=False) cfg = GatewayConfig( llm={ "provider": "volcengine", "model": "deepseek-v4-pro", "api_key": "fake", "base_url": "https://volcengine.example/api/v3", }, llm_ensemble={ "enabled": True, "selection_mode": "custom_b5", "candidates": [ {"provider": "volcengine", "model": "doubao-2.0-pro"}, {"provider": "openrouter", "model": "z-ai/glm-5.2"}, ], }, ) ready, reason = custom_b5_lineup_ready(cfg) assert ready is False assert reason == "missing_credential:openrouter" monkeypatch.setenv("OPENROUTER_API_KEY", "or-key") ready, reason = custom_b5_lineup_ready(cfg) assert ready is True assert reason == ""