wehub-resource-sync
547 lines
18 KiB
Python
547 lines
18 KiB
Python
"""Unit tests for the capability_gate routing stage and its catalog facts.
|
|
|
|
The gate acts only on DEFINITE catalog signals: an unknown model, a
|
|
synthesized entry, or the anthropic/ollama flag-gated empty capabilities
|
|
must produce no action ("never act on ignorance"). The composed default —
|
|
no capability data at all — is byte-identical to the pre-gate pipeline and
|
|
is pinned separately by ``test_routing_policy_parity.py``.
|
|
|
|
All model/tier/provider names are synthetic dummy data; the catalog is
|
|
populated in-process (no network, no credentials).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import copy
|
|
from collections.abc import Iterator
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from opensquilla.engine.pipeline import TurnContext
|
|
from opensquilla.engine.routing import (
|
|
CapabilityGateAction,
|
|
PolicyInputs,
|
|
RoutingDecision,
|
|
RoutingPolicyEngine,
|
|
TierCapability,
|
|
capability_gate,
|
|
record_capability_gate_trail,
|
|
)
|
|
from opensquilla.engine.steps import squilla_router as squilla_router_step
|
|
from opensquilla.engine.steps.squilla_router import _tier_capability_facts
|
|
from opensquilla.gateway.config import GatewayConfig
|
|
from opensquilla.provider.model_catalog import ModelCatalog, set_shared_catalog
|
|
|
|
VALID_TIERS = ["c0", "c1", "c2", "c3"]
|
|
|
|
TIERS = {
|
|
"c0": {"model": "dummy-nano-1"},
|
|
"c1": {"model": "dummy-mini-1"},
|
|
"c2": {"model": "dummy-pro-1"},
|
|
"c3": {"model": "dummy-max-1"},
|
|
}
|
|
|
|
|
|
def caps(**overrides: TierCapability) -> dict[str, TierCapability]:
|
|
facts = {name: TierCapability() for name in VALID_TIERS}
|
|
facts.update(overrides)
|
|
return facts
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# vision walk-up
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_vision_walk_up_to_nearest_vision_tier() -> None:
|
|
result = capability_gate(
|
|
"c0",
|
|
valid_tiers=VALID_TIERS,
|
|
tier_capabilities=caps(
|
|
c0=TierCapability(supports_vision=False),
|
|
c2=TierCapability(supports_vision=True),
|
|
c3=TierCapability(supports_vision=True),
|
|
),
|
|
turn_has_image=True,
|
|
material_tokens=100,
|
|
)
|
|
assert result.tier == "c2" # nearest definite vision tier, not the top
|
|
assert result.actions == (CapabilityGateAction("vision_walk_up", "c0", "c2"),)
|
|
|
|
|
|
def test_vision_unknown_caps_no_op() -> None:
|
|
# supports_vision=None covers unknown models, synthesized entries, and
|
|
# the anthropic/ollama flag-gated empty capabilities: no action.
|
|
result = capability_gate(
|
|
"c0",
|
|
valid_tiers=VALID_TIERS,
|
|
tier_capabilities=caps(c2=TierCapability(supports_vision=True)),
|
|
turn_has_image=True,
|
|
material_tokens=100,
|
|
)
|
|
assert (result.tier, result.actions) == ("c0", ())
|
|
|
|
|
|
def test_vision_no_definite_target_no_op() -> None:
|
|
# The working tier definitely lacks vision but no tier above is
|
|
# definitely vision-capable: moving anyway would be acting on ignorance.
|
|
result = capability_gate(
|
|
"c1",
|
|
valid_tiers=VALID_TIERS,
|
|
tier_capabilities=caps(
|
|
c1=TierCapability(supports_vision=False),
|
|
c2=TierCapability(supports_vision=False),
|
|
),
|
|
turn_has_image=True,
|
|
material_tokens=100,
|
|
)
|
|
assert (result.tier, result.actions) == ("c1", ())
|
|
|
|
|
|
def test_vision_rule_requires_image_turn() -> None:
|
|
result = capability_gate(
|
|
"c0",
|
|
valid_tiers=VALID_TIERS,
|
|
tier_capabilities=caps(
|
|
c0=TierCapability(supports_vision=False),
|
|
c2=TierCapability(supports_vision=True),
|
|
),
|
|
turn_has_image=False,
|
|
material_tokens=100,
|
|
)
|
|
assert (result.tier, result.actions) == ("c0", ())
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# context walk-up
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_context_walk_up_to_first_fitting_tier() -> None:
|
|
result = capability_gate(
|
|
"c0",
|
|
valid_tiers=VALID_TIERS,
|
|
tier_capabilities=caps(
|
|
c0=TierCapability(context_window=8_000),
|
|
c2=TierCapability(context_window=200_000),
|
|
),
|
|
turn_has_image=False,
|
|
material_tokens=30_000,
|
|
)
|
|
assert result.tier == "c2" # c1's window is unknown, so it cannot qualify
|
|
assert result.actions == (CapabilityGateAction("context_walk_up", "c0", "c2"),)
|
|
|
|
|
|
def test_context_unknown_window_no_op() -> None:
|
|
result = capability_gate(
|
|
"c0",
|
|
valid_tiers=VALID_TIERS,
|
|
tier_capabilities=caps(c2=TierCapability(context_window=200_000)),
|
|
turn_has_image=False,
|
|
material_tokens=500_000,
|
|
)
|
|
assert (result.tier, result.actions) == ("c0", ())
|
|
|
|
|
|
def test_context_fitting_window_no_op() -> None:
|
|
result = capability_gate(
|
|
"c0",
|
|
valid_tiers=VALID_TIERS,
|
|
tier_capabilities=caps(c0=TierCapability(context_window=64_000)),
|
|
turn_has_image=False,
|
|
material_tokens=30_000,
|
|
)
|
|
assert (result.tier, result.actions) == ("c0", ())
|
|
|
|
|
|
def test_context_top_tier_saturation_when_nothing_fits() -> None:
|
|
result = capability_gate(
|
|
"c0",
|
|
valid_tiers=VALID_TIERS,
|
|
tier_capabilities=caps(
|
|
c0=TierCapability(context_window=8_000),
|
|
c3=TierCapability(context_window=100_000),
|
|
),
|
|
turn_has_image=False,
|
|
material_tokens=500_000,
|
|
)
|
|
assert result.tier == "c3"
|
|
assert result.actions == (CapabilityGateAction("context_walk_up", "c0", "c3"),)
|
|
|
|
|
|
def test_context_already_top_tier_no_op() -> None:
|
|
result = capability_gate(
|
|
"c3",
|
|
valid_tiers=VALID_TIERS,
|
|
tier_capabilities=caps(c3=TierCapability(context_window=8_000)),
|
|
turn_has_image=False,
|
|
material_tokens=30_000,
|
|
)
|
|
assert (result.tier, result.actions) == ("c3", ())
|
|
|
|
|
|
def test_vision_then_context_walk_up_compose() -> None:
|
|
result = capability_gate(
|
|
"c0",
|
|
valid_tiers=VALID_TIERS,
|
|
tier_capabilities=caps(
|
|
c0=TierCapability(supports_vision=False),
|
|
c1=TierCapability(supports_vision=True, context_window=8_000),
|
|
c3=TierCapability(supports_vision=True, context_window=200_000),
|
|
),
|
|
turn_has_image=True,
|
|
material_tokens=30_000,
|
|
)
|
|
assert result.tier == "c3"
|
|
assert result.actions == (
|
|
CapabilityGateAction("vision_walk_up", "c0", "c1"),
|
|
CapabilityGateAction("context_walk_up", "c1", "c3"),
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# no-signal defaults + trail recording
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_no_capability_data_is_strict_no_op() -> None:
|
|
for tier_capabilities in (None, {}):
|
|
result = capability_gate(
|
|
"c0",
|
|
valid_tiers=VALID_TIERS,
|
|
tier_capabilities=tier_capabilities,
|
|
turn_has_image=True,
|
|
material_tokens=500_000,
|
|
)
|
|
assert (result.tier, result.actions) == ("c0", ())
|
|
|
|
|
|
def test_unknown_tier_is_no_op() -> None:
|
|
result = capability_gate(
|
|
"zz",
|
|
valid_tiers=VALID_TIERS,
|
|
tier_capabilities=caps(c2=TierCapability(supports_vision=True)),
|
|
turn_has_image=True,
|
|
material_tokens=100,
|
|
)
|
|
assert (result.tier, result.actions) == ("zz", ())
|
|
|
|
|
|
def test_trail_recorded_only_when_gate_acted() -> None:
|
|
extra: dict = {}
|
|
acted = capability_gate(
|
|
"c0",
|
|
valid_tiers=VALID_TIERS,
|
|
tier_capabilities=caps(
|
|
c0=TierCapability(context_window=8_000),
|
|
c2=TierCapability(context_window=200_000),
|
|
),
|
|
turn_has_image=False,
|
|
material_tokens=30_000,
|
|
)
|
|
record_capability_gate_trail(extra, acted)
|
|
assert extra["capability_gate_applied"] is True
|
|
assert extra["routing_trail"] == [
|
|
{
|
|
"stage": "capability_gate",
|
|
"rule": "context_walk_up",
|
|
"from_tier": "c0",
|
|
"to_tier": "c2",
|
|
}
|
|
]
|
|
|
|
untouched: dict = {}
|
|
idle = capability_gate(
|
|
"c0",
|
|
valid_tiers=VALID_TIERS,
|
|
tier_capabilities=None,
|
|
turn_has_image=True,
|
|
material_tokens=500_000,
|
|
)
|
|
record_capability_gate_trail(untouched, idle)
|
|
assert untouched == {} # no keys at all on a no-op turn
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# engine composition
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _engine_inputs(**overrides: object) -> PolicyInputs:
|
|
values: dict = {
|
|
"decision": RoutingDecision(
|
|
tier="c0", model="dummy-nano-1", confidence=0.9, source="v4_phase3"
|
|
),
|
|
"message": "please summarize this short note",
|
|
"router_cfg": SimpleNamespace(
|
|
default_tier="c1",
|
|
confidence_threshold=0.5,
|
|
confidence_high_tier_margin=0.05,
|
|
complaint_upgrade_enabled=True,
|
|
complaint_upgrade_steps=1,
|
|
complaint_upgrade_max_chars=160,
|
|
kv_cache_anti_downgrade_enabled=True,
|
|
kv_cache_anti_downgrade_window_seconds=600,
|
|
),
|
|
"tiers": TIERS,
|
|
"valid_tiers": VALID_TIERS,
|
|
"routing_history": None,
|
|
"extra": {},
|
|
"thinking_mode": "T1",
|
|
"prompt_policy": "P1",
|
|
"history_strategy": True,
|
|
"material_estimated_tokens": 100,
|
|
"context_window_tokens": 200_000,
|
|
"now": 10_000.0,
|
|
}
|
|
values.update(overrides)
|
|
return PolicyInputs(**values)
|
|
|
|
|
|
def test_engine_gate_runs_before_bind_and_reconciles() -> None:
|
|
extra: dict = {}
|
|
result = RoutingPolicyEngine().run(
|
|
_engine_inputs(
|
|
decision=RoutingDecision(
|
|
tier="c1", model="dummy-mini-1", confidence=0.9, source="v4_phase3"
|
|
),
|
|
extra=extra,
|
|
turn_has_image=True,
|
|
tier_capabilities=caps(
|
|
c1=TierCapability(supports_vision=False),
|
|
c2=TierCapability(supports_vision=True),
|
|
),
|
|
)
|
|
)
|
|
assert result.decision.tier == "c2"
|
|
assert result.decision.model == "dummy-pro-1" # bind rebound to the gated tier
|
|
assert extra["final_tier"] == "c2"
|
|
assert extra["final_route_class"] == "R2"
|
|
assert extra["capability_gate_applied"] is True
|
|
assert extra["routing_trail"] == [
|
|
{
|
|
"stage": "capability_gate",
|
|
"rule": "vision_walk_up",
|
|
"from_tier": "c1",
|
|
"to_tier": "c2",
|
|
}
|
|
]
|
|
assert result.thinking_mode == "T2" # reconciled against the gated final tier
|
|
|
|
|
|
def test_engine_default_inputs_match_pre_gate_pipeline() -> None:
|
|
baseline_extra: dict = {}
|
|
baseline = RoutingPolicyEngine().run(_engine_inputs(extra=baseline_extra))
|
|
gated_extra: dict = {}
|
|
gated = RoutingPolicyEngine().run(
|
|
_engine_inputs(
|
|
extra=gated_extra,
|
|
turn_has_image=False,
|
|
tier_capabilities=None,
|
|
)
|
|
)
|
|
assert gated.decision == baseline.decision
|
|
assert gated_extra == baseline_extra
|
|
assert "capability_gate_applied" not in gated_extra
|
|
assert "routing_trail" not in gated_extra
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# catalog facts gathering (never act on ignorance, at the source)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture()
|
|
def facts_catalog() -> Iterator[ModelCatalog]:
|
|
catalog = ModelCatalog()
|
|
catalog._populate_from_data(
|
|
[
|
|
{
|
|
"id": "dummy-vis-live-1",
|
|
"context_length": 200_000,
|
|
"supported_parameters": ["tools"],
|
|
"architecture": {"input_modalities": ["text", "image"]},
|
|
},
|
|
{
|
|
"id": "dummy-small-live-1",
|
|
"context_length": 8_000,
|
|
"supported_parameters": ["reasoning"],
|
|
"architecture": {"input_modalities": ["text"]},
|
|
},
|
|
]
|
|
)
|
|
catalog.set_user_overrides(
|
|
{"anthroprov/dummy-anthropic-x": {"supports_vision": True}}
|
|
)
|
|
set_shared_catalog(catalog)
|
|
try:
|
|
yield catalog
|
|
finally:
|
|
set_shared_catalog(None)
|
|
|
|
|
|
def test_facts_definite_signals_from_catalog(facts_catalog: ModelCatalog) -> None:
|
|
tiers = {
|
|
"c0": {"model": "dummy-small-live-1"},
|
|
"c1": {"model": "dummy-mini-unknown-1"},
|
|
"c2": {"model": "dummy-vis-live-1"},
|
|
}
|
|
facts = _tier_capability_facts(tiers, ["c0", "c1", "c2"], "mainprov")
|
|
assert facts["c0"] == TierCapability(supports_vision=False, context_window=8_000)
|
|
assert facts["c2"] == TierCapability(supports_vision=True, context_window=200_000)
|
|
|
|
|
|
def test_facts_synthesized_entry_gives_no_signal(facts_catalog: ModelCatalog) -> None:
|
|
facts = _tier_capability_facts(
|
|
{"c1": {"model": "dummy-mini-unknown-1"}}, ["c1"], "mainprov"
|
|
)
|
|
assert facts["c1"] == TierCapability(supports_vision=None, context_window=None)
|
|
|
|
|
|
def test_facts_user_override_window_counts_as_definite(facts_catalog: ModelCatalog) -> None:
|
|
# An operator-declared [models.*] context_window is knowledge, not an
|
|
# estimate: the gate may act on it exactly like a catalog-sourced window.
|
|
facts_catalog.set_user_overrides(
|
|
{"mainprov/dummy-pinned-window-1": {"context_window": 64_000}}
|
|
)
|
|
facts = _tier_capability_facts(
|
|
{"c1": {"model": "dummy-pinned-window-1"}}, ["c1"], "mainprov"
|
|
)
|
|
assert facts["c1"].context_window == 64_000
|
|
|
|
|
|
def test_facts_anthropic_flag_gated_caps_stay_unknown(facts_catalog: ModelCatalog) -> None:
|
|
# The user-override layer claims vision for this model. Under a normal
|
|
# provider that is a definite signal; under anthropic the capabilities
|
|
# are flag-gated to the empty ModelCapabilities() for one release, so
|
|
# the gate must treat the same claim as unknown.
|
|
overridden = _tier_capability_facts(
|
|
{"c2": {"model": "dummy-anthropic-x", "provider": "anthroprov"}},
|
|
["c2"],
|
|
"mainprov",
|
|
)
|
|
assert overridden["c2"].supports_vision is True
|
|
|
|
anthropic_facts = _tier_capability_facts(
|
|
{"c2": {"model": "dummy-anthropic-x", "provider": "anthropic"}},
|
|
["c2"],
|
|
"mainprov",
|
|
)
|
|
assert anthropic_facts["c2"] == TierCapability(
|
|
supports_vision=None, context_window=None
|
|
)
|
|
|
|
|
|
def test_facts_blank_model_gives_no_signal(facts_catalog: ModelCatalog) -> None:
|
|
facts = _tier_capability_facts({"c0": {"model": ""}}, ["c0"], "mainprov")
|
|
assert facts["c0"] == TierCapability()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# step-level: context walk-up through apply_squilla_router
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class _FixedStrategy:
|
|
def __init__(self, tier: str, confidence: float, extra: dict) -> None:
|
|
self.tier = tier
|
|
self.confidence = confidence
|
|
self.extra = extra
|
|
|
|
async def classify(
|
|
self,
|
|
message: str,
|
|
valid_tiers: list[str],
|
|
routing_history: list[dict] | None = None,
|
|
) -> tuple[str, float, str, dict]:
|
|
return self.tier, self.confidence, "v4_phase3", copy.deepcopy(self.extra)
|
|
|
|
|
|
def _run_step(
|
|
config: GatewayConfig,
|
|
strategy: object,
|
|
*,
|
|
message: str = "please summarize this short note",
|
|
metadata: dict | None = None,
|
|
) -> TurnContext:
|
|
sr = squilla_router_step
|
|
sr._history_store.clear()
|
|
sr._strategy = None
|
|
sr._strategy_key = None
|
|
ctx = TurnContext(
|
|
message=message,
|
|
session_key="agent:capgate:main",
|
|
config=config,
|
|
provider=None,
|
|
model=config.llm.model,
|
|
tool_defs=[],
|
|
system_prompt="system",
|
|
attachments=[],
|
|
metadata=metadata or {},
|
|
)
|
|
original = sr._get_strategy
|
|
sr._get_strategy = lambda _config: strategy # type: ignore[assignment]
|
|
try:
|
|
asyncio.run(sr.apply_squilla_router(ctx))
|
|
finally:
|
|
sr._get_strategy = original # type: ignore[assignment]
|
|
sr._history_store.clear()
|
|
return ctx
|
|
|
|
|
|
def _gate_config() -> GatewayConfig:
|
|
config = GatewayConfig()
|
|
config.llm.provider = "mainprov"
|
|
config.llm.model = "dummy-base-model"
|
|
config.squilla_router.enabled = True
|
|
config.squilla_router.rollout_phase = "full"
|
|
config.squilla_router.tiers = {
|
|
"c0": {"model": "dummy-small-live-1"},
|
|
"c1": {"model": "dummy-mini-unknown-1"},
|
|
"c2": {"model": "dummy-vis-live-1"},
|
|
"c3": {"model": "dummy-max-unknown-1"},
|
|
}
|
|
return config
|
|
|
|
|
|
def test_step_context_walk_up_end_to_end(facts_catalog: ModelCatalog) -> None:
|
|
# 10k material tokens: above c0's definite 8k window, below the 25k
|
|
# large-context floor — only the capability gate can move this turn.
|
|
ctx = _run_step(
|
|
_gate_config(),
|
|
_FixedStrategy("c0", 0.9, {"route_class": "R0", "thinking_mode": "T0"}),
|
|
metadata={"material_estimated_tokens": 10_000},
|
|
)
|
|
assert ctx.metadata["routed_tier"] == "c2"
|
|
assert ctx.metadata["routed_model"] == "dummy-vis-live-1"
|
|
assert ctx.metadata["routing_extra"]["capability_gate_applied"] is True
|
|
assert ctx.metadata["routing_extra"]["routing_trail"] == [
|
|
{
|
|
"stage": "capability_gate",
|
|
"rule": "context_walk_up",
|
|
"from_tier": "c0",
|
|
"to_tier": "c2",
|
|
}
|
|
]
|
|
assert ctx.metadata["routing_extra"]["final_tier"] == "c2"
|
|
|
|
|
|
def test_step_no_gate_keys_when_catalog_unknown(facts_catalog: ModelCatalog) -> None:
|
|
config = _gate_config()
|
|
config.squilla_router.tiers = {
|
|
"c0": {"model": "dummy-unknown-a"},
|
|
"c1": {"model": "dummy-unknown-b"},
|
|
"c2": {"model": "dummy-unknown-c"},
|
|
"c3": {"model": "dummy-unknown-d"},
|
|
}
|
|
ctx = _run_step(
|
|
config,
|
|
_FixedStrategy("c0", 0.9, {"route_class": "R0", "thinking_mode": "T0"}),
|
|
metadata={"material_estimated_tokens": 10_000},
|
|
)
|
|
assert ctx.metadata["routed_tier"] == "c0"
|
|
assert "capability_gate_applied" not in ctx.metadata["routing_extra"]
|
|
assert "routing_trail" not in ctx.metadata["routing_extra"]
|