wehub-resource-sync
1331 lines
43 KiB
Python
1331 lines
43 KiB
Python
import logging
|
|
|
|
import pytest
|
|
|
|
from opensquilla.engine.pipeline import TurnContext
|
|
from opensquilla.engine.steps import squilla_router as squilla_router_step
|
|
from opensquilla.engine.steps.squilla_router import apply_squilla_router
|
|
from opensquilla.gateway.config import GatewayConfig
|
|
from opensquilla.squilla_router.v4_phase3 import V4Phase3Strategy
|
|
|
|
|
|
class FakeStrategy:
|
|
def __init__(self, tier: str, confidence: float, extra: dict) -> None:
|
|
self.tier = tier
|
|
self.confidence = confidence
|
|
self.extra = extra
|
|
self.calls = 0
|
|
self.messages: list[str] = []
|
|
|
|
async def classify(
|
|
self,
|
|
message: str,
|
|
valid_tiers: list[str],
|
|
routing_history: list[dict] | None = None,
|
|
) -> tuple[str, float, str, dict]:
|
|
self.calls += 1
|
|
self.messages.append(message)
|
|
assert self.tier in valid_tiers
|
|
return self.tier, self.confidence, "v4_phase3", dict(self.extra)
|
|
|
|
|
|
class ContextAwareFakeStrategy(FakeStrategy):
|
|
def __init__(self, tier: str, confidence: float, extra: dict) -> None:
|
|
super().__init__(tier, confidence, extra)
|
|
self.contexts: list[dict] = []
|
|
|
|
async def classify(
|
|
self,
|
|
message: str,
|
|
valid_tiers: list[str],
|
|
routing_history: list[dict] | None = None,
|
|
prev_assistant_text: str | None = None,
|
|
prev_assistant_usage: dict | None = None,
|
|
history_user_texts: list[str] | None = None,
|
|
flags_text_override: str | None = None,
|
|
) -> tuple[str, float, str, dict]:
|
|
self.calls += 1
|
|
self.messages.append(message)
|
|
self.contexts.append(
|
|
{
|
|
"routing_history": [dict(entry) for entry in routing_history or []],
|
|
"prev_assistant_text": prev_assistant_text,
|
|
"prev_assistant_usage": dict(prev_assistant_usage or {}),
|
|
"history_user_texts": list(history_user_texts or []),
|
|
"flags_text_override": flags_text_override,
|
|
}
|
|
)
|
|
assert self.tier in valid_tiers
|
|
return self.tier, self.confidence, "v4_phase3", dict(self.extra)
|
|
|
|
|
|
class ExplodingV4Strategy:
|
|
def __init__(self, *args, **kwargs) -> None:
|
|
raise RuntimeError(
|
|
"failed to initialize V4 Phase 3 router: DLL load failed while importing "
|
|
"onnxruntime_pybind11_state"
|
|
)
|
|
|
|
|
|
class MacLibompExplodingV4Strategy:
|
|
def __init__(self, *args, **kwargs) -> None:
|
|
raise RuntimeError(
|
|
"failed to initialize V4 Phase 3 router: dlopen("
|
|
".../lightgbm/lib/lib_lightgbm.dylib, 0x0006): Library not loaded: "
|
|
"@rpath/libomp.dylib Referenced from: .../lib_lightgbm.dylib"
|
|
)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def reset_squilla_router_state(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
squilla_router_step._history_store.clear()
|
|
squilla_router_step._strategy = None
|
|
squilla_router_step._strategy_key = None
|
|
squilla_router_step._router_runtime_warning_emitted = False
|
|
yield
|
|
squilla_router_step._history_store.clear()
|
|
squilla_router_step._strategy = None
|
|
squilla_router_step._strategy_key = None
|
|
squilla_router_step._router_runtime_warning_emitted = False
|
|
monkeypatch.undo()
|
|
|
|
|
|
def make_context(
|
|
message: str,
|
|
*,
|
|
rollout_phase: str = "full",
|
|
session_key: str = "test-session",
|
|
raw_message: str | None = None,
|
|
attachments: list[dict] | None = None,
|
|
) -> TurnContext:
|
|
# Pin the packaged openrouter ladder: these tests assert tier moves by
|
|
# their distinct per-tier models, which the tokenrhythm default's uniform
|
|
# synthesized ladder cannot express.
|
|
config = GatewayConfig(llm={"provider": "openrouter"})
|
|
config.squilla_router.rollout_phase = rollout_phase
|
|
return TurnContext(
|
|
message=message,
|
|
session_key=session_key,
|
|
config=config,
|
|
provider=None,
|
|
model=config.llm.model,
|
|
tool_defs=[],
|
|
system_prompt="system",
|
|
raw_message=raw_message,
|
|
attachments=attachments or [],
|
|
)
|
|
|
|
|
|
def require_runtime_router() -> None:
|
|
try:
|
|
V4Phase3Strategy(require_router_runtime=True)
|
|
except Exception as exc:
|
|
pytest.skip(f"V4 model router runtime unavailable: {exc}")
|
|
|
|
|
|
def fake_strategy(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tier: str,
|
|
confidence: float,
|
|
extra: dict,
|
|
) -> FakeStrategy:
|
|
strategy = FakeStrategy(tier, confidence, extra)
|
|
monkeypatch.setattr(squilla_router_step, "_get_strategy", lambda _config: strategy)
|
|
return strategy
|
|
|
|
|
|
def context_aware_fake_strategy(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tier: str,
|
|
confidence: float,
|
|
extra: dict,
|
|
) -> ContextAwareFakeStrategy:
|
|
strategy = ContextAwareFakeStrategy(tier, confidence, extra)
|
|
monkeypatch.setattr(squilla_router_step, "_get_strategy", lambda _config: strategy)
|
|
return strategy
|
|
|
|
|
|
def test_vision_followup_gate_default_timeout_allows_real_provider_latency() -> None:
|
|
config = GatewayConfig()
|
|
|
|
assert config.squilla_router.vision_followup_gate_timeout_seconds >= 10.0
|
|
|
|
|
|
def test_vision_followup_gate_default_output_budget_handles_reasoning_models() -> None:
|
|
config = GatewayConfig()
|
|
|
|
assert config.squilla_router.vision_followup_gate_max_output_tokens >= 512
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_full_rollout_applies_routed_model_thinking_and_p0_prompt(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
fake_strategy(
|
|
monkeypatch,
|
|
"c1",
|
|
0.91,
|
|
{
|
|
"route_class": "R1",
|
|
"thinking_mode": "T1",
|
|
"prompt_policy": "P0",
|
|
},
|
|
)
|
|
ctx = make_context("Summarize this short note.")
|
|
baseline_model = ctx.model
|
|
|
|
routed = await apply_squilla_router(ctx)
|
|
|
|
assert routed.model == "deepseek/deepseek-v4-pro"
|
|
assert routed.metadata["routed_tier"] == "c1"
|
|
assert routed.metadata["routed_model"] == "deepseek/deepseek-v4-pro"
|
|
assert routed.metadata["routing_applied"] is True
|
|
assert routed.metadata["applied_model"] == "deepseek/deepseek-v4-pro"
|
|
assert routed.metadata["baseline_model"] == baseline_model
|
|
assert routed.metadata["routing_confidence"] == 0.91
|
|
assert routed.metadata["routing_source"] == "v4_phase3"
|
|
assert "savings_pct" in routed.metadata
|
|
assert "savings_max_price_per_m" in routed.metadata
|
|
assert "savings_routed_price_per_m" in routed.metadata
|
|
assert routed.metadata["thinking_mode"] == "T1"
|
|
assert routed.metadata["thinking_requested"] is True
|
|
assert routed.metadata["thinking_level"] == "low"
|
|
assert routed.metadata["prompt_policy"] == "P0"
|
|
assert "[RESPONSE_POLICY: Answer directly" in routed.message
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_router_records_lower_text_tier_fallback_chain(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
fake_strategy(
|
|
monkeypatch,
|
|
"c3",
|
|
0.91,
|
|
{
|
|
"route_class": "R3",
|
|
"thinking_mode": "T1",
|
|
"prompt_policy": "P0",
|
|
},
|
|
)
|
|
|
|
routed = await apply_squilla_router(make_context("Solve a difficult architecture problem."))
|
|
|
|
assert routed.metadata["routed_tier"] == "c3"
|
|
assert [item["tier"] for item in routed.metadata["router_fallback_chain"]] == [
|
|
"c2",
|
|
"c1",
|
|
"c0",
|
|
]
|
|
assert [item["model"] for item in routed.metadata["router_fallback_chain"]] == [
|
|
"z-ai/glm-5.2",
|
|
"deepseek/deepseek-v4-pro",
|
|
"deepseek/deepseek-v4-flash",
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_router_reports_provider_state_loss_without_changing_route(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
fake_strategy(
|
|
monkeypatch,
|
|
"c1",
|
|
0.91,
|
|
{
|
|
"route_class": "R1",
|
|
"thinking_mode": "T1",
|
|
"prompt_policy": "P0",
|
|
},
|
|
)
|
|
ctx = make_context("Continue the long task.")
|
|
ctx.metadata["session_context_states"] = [
|
|
{
|
|
"provider": "anthropic",
|
|
"model": "claude-sonnet-4-6",
|
|
"state_kind": "anthropic_compaction_block",
|
|
"valid": True,
|
|
"portable": False,
|
|
},
|
|
{
|
|
"provider": "portable",
|
|
"model": "",
|
|
"state_kind": "structured_summary_v1",
|
|
"valid": True,
|
|
"portable": True,
|
|
},
|
|
]
|
|
|
|
routed = await apply_squilla_router(ctx)
|
|
|
|
assert routed.model == "deepseek/deepseek-v4-pro"
|
|
diagnostic = routed.metadata["provider_state_continuity"]
|
|
assert diagnostic["decision"] == "use_portable_fallback"
|
|
assert diagnostic["provider_state_loss_risk"] is True
|
|
assert diagnostic["candidate_provider"] == "openrouter"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_router_continuity_diagnostic_ignores_expired_provider_state(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
fake_strategy(
|
|
monkeypatch,
|
|
"c1",
|
|
0.91,
|
|
{
|
|
"route_class": "R1",
|
|
"thinking_mode": "T1",
|
|
"prompt_policy": "P0",
|
|
},
|
|
)
|
|
ctx = make_context("Continue the long task.")
|
|
ctx.metadata["session_context_states"] = [
|
|
{
|
|
"provider": "anthropic",
|
|
"model": "claude-sonnet-4-6",
|
|
"state_kind": "anthropic_compaction_block",
|
|
"created_at": 100,
|
|
"expires_at": 150,
|
|
"valid": True,
|
|
"portable": False,
|
|
},
|
|
{
|
|
"provider": "portable",
|
|
"model": "",
|
|
"state_kind": "structured_summary_v1",
|
|
"created_at": 90,
|
|
"valid": True,
|
|
"portable": True,
|
|
},
|
|
]
|
|
|
|
routed = await apply_squilla_router(ctx)
|
|
|
|
diagnostic = routed.metadata["provider_state_continuity"]
|
|
assert diagnostic["decision"] == "use_portable_fallback"
|
|
assert diagnostic["provider_state_loss_risk"] is False
|
|
assert diagnostic["active_state_provider"] is None
|
|
assert diagnostic["portable_fallback_available"] is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_p2_prompt_hint_is_recorded_but_not_injected(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
fake_strategy(
|
|
monkeypatch,
|
|
"c3",
|
|
0.97,
|
|
{
|
|
"route_class": "R3",
|
|
"thinking_mode": "T3",
|
|
"prompt_policy": "P2",
|
|
"prompt_hint": "Use a careful plan before answering.",
|
|
},
|
|
)
|
|
ctx = make_context("Plan a risky multi-step migration.")
|
|
|
|
routed = await apply_squilla_router(ctx)
|
|
|
|
assert routed.model == "anthropic/claude-opus-4.8"
|
|
assert routed.metadata["routed_tier"] == "c3"
|
|
assert routed.metadata["thinking_level"] == "high"
|
|
assert routed.metadata["prompt_policy"] == "P2"
|
|
assert routed.metadata["routing_extra"]["prompt_hint"] == "Use a careful plan before answering."
|
|
assert "[RESPONSE_POLICY:" not in routed.message
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_v4_thinking_mode_overrides_explicit_tier_thinking_level(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
fake_strategy(
|
|
monkeypatch,
|
|
"c2",
|
|
0.92,
|
|
{
|
|
"route_class": "R2",
|
|
"thinking_mode": "T2",
|
|
"prompt_policy": "P1",
|
|
},
|
|
)
|
|
ctx = make_context("Analyze this implementation path.")
|
|
|
|
routed = await apply_squilla_router(ctx)
|
|
|
|
assert routed.metadata["routed_tier"] == "c2"
|
|
assert routed.metadata["thinking_mode"] == "T2"
|
|
assert routed.metadata["thinking_requested"] is True
|
|
assert routed.metadata["thinking_level"] == "medium"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_confidence_gate_promotes_low_confidence_t0_to_default_t1_and_reconciles_thinking(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
fake_strategy(
|
|
monkeypatch,
|
|
"c0",
|
|
0.1,
|
|
{
|
|
"route_class": "R0",
|
|
"thinking_mode": "T0",
|
|
"prompt_policy": "P0",
|
|
},
|
|
)
|
|
ctx = make_context("Maybe simple, but classifier is uncertain.")
|
|
|
|
routed = await apply_squilla_router(ctx)
|
|
extra = routed.metadata["routing_extra"]
|
|
|
|
assert routed.metadata["routed_tier"] == "c1"
|
|
assert routed.model == "deepseek/deepseek-v4-pro"
|
|
assert extra["confidence_gate_applied"] is True
|
|
assert extra["base_tier"] == "c0"
|
|
assert extra["final_tier"] == "c1"
|
|
assert routed.metadata["thinking_mode"] == "T1"
|
|
assert routed.metadata["thinking_level"] == "low"
|
|
assert "[RESPONSE_POLICY: Answer directly" in routed.message
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_confidence_gate_falls_back_low_confidence_non_default_text_tier(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
fake_strategy(
|
|
monkeypatch,
|
|
"c2",
|
|
0.1,
|
|
{
|
|
"route_class": "R2",
|
|
"thinking_mode": "T2",
|
|
"prompt_policy": "P1",
|
|
},
|
|
)
|
|
ctx = make_context("Classifier is uncertain but picked an expensive tier.")
|
|
|
|
routed = await apply_squilla_router(ctx)
|
|
extra = routed.metadata["routing_extra"]
|
|
|
|
assert routed.metadata["routed_tier"] == "c1"
|
|
assert routed.model == "deepseek/deepseek-v4-pro"
|
|
assert extra["confidence_gate_applied"] is True
|
|
assert extra["pre_confidence_tier"] == "c2"
|
|
assert extra["final_tier"] == "c1"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_confidence_gate_keeps_near_threshold_high_tier(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
fake_strategy(
|
|
monkeypatch,
|
|
"c3",
|
|
0.49,
|
|
{
|
|
"route_class": "R3",
|
|
"thinking_mode": "T3",
|
|
"prompt_policy": "P2",
|
|
},
|
|
)
|
|
ctx = make_context("Classifier is near threshold but chose a high-risk tier.")
|
|
|
|
routed = await apply_squilla_router(ctx)
|
|
extra = routed.metadata["routing_extra"]
|
|
|
|
assert routed.metadata["routed_tier"] == "c3"
|
|
assert extra["confidence_gate_applied"] is False
|
|
assert extra["pre_confidence_tier"] == "c3"
|
|
assert extra["final_tier"] == "c3"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_large_material_estimate_floors_low_router_tier(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
fake_strategy(monkeypatch, "c1", 0.91, {"route_class": "R1"})
|
|
ctx = make_context("Please process the attached pasted text.")
|
|
ctx.metadata["input_normalization"] = {
|
|
"guard_action": "generated_text_attachment",
|
|
"material_estimated_tokens": 45_000,
|
|
}
|
|
ctx.metadata["material_estimated_tokens"] = 45_000
|
|
|
|
routed = await apply_squilla_router(ctx)
|
|
|
|
assert routed.metadata["routed_tier"] == "c2"
|
|
assert routed.metadata["routing_source"] == "large_context_floor"
|
|
assert routed.metadata["large_context_floor_from_tier"] == "c1"
|
|
assert routed.metadata["large_context_material_tokens"] == 45_000
|
|
assert routed.metadata["routing_extra"]["final_tier"] == "c2"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_large_material_ratio_floors_low_router_tier_to_t3(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
fake_strategy(monkeypatch, "c1", 0.91, {"route_class": "R1"})
|
|
ctx = make_context("Please process the attached pasted text.")
|
|
object.__setattr__(ctx.config.squilla_router, "context_window_tokens", 100_000)
|
|
ctx.metadata["input_normalization"] = {
|
|
"guard_action": "generated_text_attachment",
|
|
"material_estimated_tokens": 40_000,
|
|
}
|
|
|
|
routed = await apply_squilla_router(ctx)
|
|
|
|
assert routed.metadata["routed_tier"] == "c3"
|
|
assert routed.metadata["routing_source"] == "large_context_floor"
|
|
assert routed.metadata["large_context_floor_from_tier"] == "c1"
|
|
assert routed.metadata["large_context_material_tokens"] == 40_000
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_anti_downgrade_keeps_recent_higher_tier_despite_confidence_gate(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
ctx1 = make_context("Hard first turn.", session_key="test-confidence-history")
|
|
fake_strategy(
|
|
monkeypatch,
|
|
"c2",
|
|
0.9,
|
|
{
|
|
"route_class": "R2",
|
|
"thinking_mode": "T2",
|
|
"prompt_policy": "P1",
|
|
},
|
|
)
|
|
routed1 = await apply_squilla_router(ctx1)
|
|
assert routed1.metadata["routed_tier"] == "c2"
|
|
|
|
fake_strategy(
|
|
monkeypatch,
|
|
"c0",
|
|
0.1,
|
|
{
|
|
"route_class": "R0",
|
|
"thinking_mode": "T0",
|
|
"prompt_policy": "P0",
|
|
},
|
|
)
|
|
ctx2 = make_context("Uncertain follow-up.", session_key="test-confidence-history")
|
|
|
|
routed2 = await apply_squilla_router(ctx2)
|
|
extra = routed2.metadata["routing_extra"]
|
|
|
|
assert routed2.metadata["routed_tier"] == "c2"
|
|
assert routed2.model == "z-ai/glm-5.2"
|
|
assert extra["confidence_gate_applied"] is True
|
|
assert extra["pre_confidence_tier"] == "c0"
|
|
assert extra["final_tier"] == "c2"
|
|
assert extra["anti_downgrade_applied"] is True
|
|
assert extra["previous_tier"] == "c2"
|
|
|
|
fake_strategy(
|
|
monkeypatch,
|
|
"c1",
|
|
0.9,
|
|
{
|
|
"route_class": "R1",
|
|
"thinking_mode": "T1",
|
|
"prompt_policy": "P1",
|
|
},
|
|
)
|
|
ctx3 = make_context("Normal follow-up.", session_key="test-confidence-history")
|
|
|
|
routed3 = await apply_squilla_router(ctx3)
|
|
extra3 = routed3.metadata["routing_extra"]
|
|
|
|
assert routed3.metadata["routed_tier"] == "c2"
|
|
assert routed3.model == "z-ai/glm-5.2"
|
|
assert extra3["confidence_gate_applied"] is False
|
|
assert extra3["anti_downgrade_applied"] is True
|
|
assert extra3["previous_tier"] == "c2"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_anti_downgrade_uses_previous_turn_not_window_highest(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
session_key = "test-previous-not-highest"
|
|
fake_strategy(
|
|
monkeypatch,
|
|
"c3",
|
|
0.9,
|
|
{
|
|
"route_class": "R3",
|
|
"thinking_mode": "T3",
|
|
"prompt_policy": "P2",
|
|
},
|
|
)
|
|
routed1 = await apply_squilla_router(make_context("Very hard turn.", session_key=session_key))
|
|
assert routed1.metadata["routed_tier"] == "c3"
|
|
|
|
ctx2 = make_context("Less hard turn.", session_key=session_key)
|
|
ctx2.config.squilla_router.kv_cache_anti_downgrade_enabled = False
|
|
fake_strategy(
|
|
monkeypatch,
|
|
"c2",
|
|
0.9,
|
|
{
|
|
"route_class": "R2",
|
|
"thinking_mode": "T2",
|
|
"prompt_policy": "P1",
|
|
},
|
|
)
|
|
routed2 = await apply_squilla_router(ctx2)
|
|
assert routed2.metadata["routed_tier"] == "c2"
|
|
|
|
ctx3 = make_context("Easy follow-up.", session_key=session_key)
|
|
fake_strategy(
|
|
monkeypatch,
|
|
"c1",
|
|
0.9,
|
|
{
|
|
"route_class": "R1",
|
|
"thinking_mode": "T1",
|
|
"prompt_policy": "P1",
|
|
},
|
|
)
|
|
routed3 = await apply_squilla_router(ctx3)
|
|
extra3 = routed3.metadata["routing_extra"]
|
|
|
|
assert routed3.metadata["routed_tier"] == "c2"
|
|
assert routed3.model == "z-ai/glm-5.2"
|
|
assert extra3["anti_downgrade_applied"] is True
|
|
assert extra3["previous_tier"] == "c2"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_anti_downgrade_keeps_previous_high_tier_without_margin_gate(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
session_key = "test-anti-downgrade-ignore-margin"
|
|
fake_strategy(
|
|
monkeypatch,
|
|
"c3",
|
|
0.95,
|
|
{
|
|
"route_class": "R3",
|
|
"thinking_mode": "T3",
|
|
"prompt_policy": "P2",
|
|
"margin": 0.99,
|
|
},
|
|
)
|
|
routed1 = await apply_squilla_router(
|
|
make_context("Architecture review.", session_key=session_key)
|
|
)
|
|
assert routed1.metadata["routed_tier"] == "c3"
|
|
|
|
fake_strategy(
|
|
monkeypatch,
|
|
"c1",
|
|
0.99,
|
|
{
|
|
"route_class": "R1",
|
|
"thinking_mode": "T1",
|
|
"prompt_policy": "P1",
|
|
"margin": 0.99,
|
|
},
|
|
)
|
|
routed2 = await apply_squilla_router(make_context("Follow-up.", session_key=session_key))
|
|
extra = routed2.metadata["routing_extra"]
|
|
|
|
assert routed2.metadata["routed_tier"] == "c3"
|
|
assert routed2.model == "anthropic/claude-opus-4.8"
|
|
assert extra["anti_downgrade_applied"] is True
|
|
assert extra["previous_tier"] == "c3"
|
|
assert extra["kv_cache_window_seconds"] == 600
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_complaint_upgrade_promotes_tier_thinking_and_blocks_compressed_prompt(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
fake_strategy(
|
|
monkeypatch,
|
|
"c1",
|
|
0.9,
|
|
{
|
|
"route_class": "R1",
|
|
"thinking_mode": "T1",
|
|
"prompt_policy": "P0",
|
|
},
|
|
)
|
|
ctx = make_context("不对，重新回答")
|
|
|
|
routed = await apply_squilla_router(ctx)
|
|
extra = routed.metadata["routing_extra"]
|
|
|
|
assert routed.metadata["routed_tier"] == "c2"
|
|
assert routed.model == "z-ai/glm-5.2"
|
|
assert extra["complaint_detected"] is True
|
|
assert extra["complaint_upgrade_applied"] is True
|
|
assert routed.metadata["thinking_mode"] == "T2"
|
|
assert routed.metadata["thinking_level"] == "medium"
|
|
assert routed.metadata["prompt_policy"] == "P1"
|
|
assert "[RESPONSE_POLICY:" not in routed.message
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_complaint_upgrade_starts_from_previous_experienced_tier(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
session_key = "test-complaint-upgrade-previous-tier"
|
|
fake_strategy(
|
|
monkeypatch,
|
|
"c2",
|
|
0.9,
|
|
{
|
|
"route_class": "R2",
|
|
"thinking_mode": "T2",
|
|
"prompt_policy": "P1",
|
|
},
|
|
)
|
|
routed1 = await apply_squilla_router(
|
|
make_context("Analyze this tricky failure.", session_key=session_key)
|
|
)
|
|
assert routed1.metadata["routed_tier"] == "c2"
|
|
|
|
fake_strategy(
|
|
monkeypatch,
|
|
"c1",
|
|
0.9,
|
|
{
|
|
"route_class": "R1",
|
|
"thinking_mode": "T1",
|
|
"prompt_policy": "P0",
|
|
},
|
|
)
|
|
routed2 = await apply_squilla_router(make_context("答非所问", session_key=session_key))
|
|
extra = routed2.metadata["routing_extra"]
|
|
|
|
assert routed2.metadata["routed_tier"] == "c3"
|
|
assert routed2.model == "anthropic/claude-opus-4.8"
|
|
assert extra["previous_tier"] == "c2"
|
|
assert extra["complaint_detected"] is True
|
|
assert extra["complaint_upgrade_applied"] is True
|
|
assert extra["anti_downgrade_applied"] is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_complaint_upgrade_uses_pre_confidence_high_tier(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
session_key = "test-complaint-upgrade-pre-confidence-tier"
|
|
fake_strategy(
|
|
monkeypatch,
|
|
"c1",
|
|
0.9,
|
|
{
|
|
"route_class": "R1",
|
|
"thinking_mode": "T1",
|
|
"prompt_policy": "P1",
|
|
},
|
|
)
|
|
await apply_squilla_router(
|
|
make_context("Compare PostgreSQL and MySQL.", session_key=session_key)
|
|
)
|
|
|
|
fake_strategy(
|
|
monkeypatch,
|
|
"c2",
|
|
0.01,
|
|
{
|
|
"route_class": "R2",
|
|
"thinking_mode": "T2",
|
|
"prompt_policy": "P1",
|
|
},
|
|
)
|
|
routed = await apply_squilla_router(
|
|
make_context("不对，太泛了，重新写。", session_key=session_key)
|
|
)
|
|
extra = routed.metadata["routing_extra"]
|
|
|
|
assert extra["confidence_gate_applied"] is True
|
|
assert extra["pre_confidence_tier"] == "c2"
|
|
assert routed.metadata["routed_tier"] == "c3"
|
|
assert extra["complaint_upgrade_applied"] is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_router_classifies_raw_semantic_input_but_injects_prompt_into_display_message(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
strategy = fake_strategy(
|
|
monkeypatch,
|
|
"c0",
|
|
0.92,
|
|
{
|
|
"route_class": "R0",
|
|
"thinking_mode": "T0",
|
|
"prompt_policy": "P0",
|
|
},
|
|
)
|
|
ctx = make_context(
|
|
"Displayed prompt wrapper",
|
|
raw_message="Summarize the underlying user input.",
|
|
)
|
|
|
|
routed = await apply_squilla_router(ctx)
|
|
|
|
assert strategy.messages == ["Summarize the underlying user input."]
|
|
assert routed.metadata["routed_tier"] == "c0"
|
|
assert routed.metadata["prompt_policy"] == "P0"
|
|
assert routed.message.startswith("Displayed prompt wrapper")
|
|
assert "Summarize the underlying user input." not in routed.message
|
|
assert "[RESPONSE_POLICY: Answer directly" in routed.message
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_router_passes_transcript_context_into_strategy(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
strategy = context_aware_fake_strategy(
|
|
monkeypatch,
|
|
"c2",
|
|
0.88,
|
|
{
|
|
"route_class": "R2",
|
|
"thinking_mode": "T2",
|
|
"prompt_policy": "P1",
|
|
},
|
|
)
|
|
ctx = make_context("Continue from the previous answer.")
|
|
ctx.metadata.update(
|
|
{
|
|
"router_prev_assistant_text": "Previous assistant answer.",
|
|
"router_prev_assistant_usage": {"output_tokens": 321},
|
|
"router_history_user_texts": ["First user question.", "Second user question."],
|
|
"router_flags_text_override": "Continue from the previous answer.",
|
|
"routing_history": [
|
|
{
|
|
"text": "First user question.",
|
|
"route_class": "R1",
|
|
"final_route_class": "R1",
|
|
"difficulty": 1.0,
|
|
"margin": 0.5,
|
|
}
|
|
],
|
|
}
|
|
)
|
|
|
|
routed = await apply_squilla_router(ctx)
|
|
|
|
assert routed.metadata["routed_tier"] == "c2"
|
|
assert strategy.messages == ["Continue from the previous answer."]
|
|
assert strategy.contexts == [
|
|
{
|
|
"routing_history": [
|
|
{
|
|
"text": "First user question.",
|
|
"route_class": "R1",
|
|
"final_route_class": "R1",
|
|
"difficulty": 1.0,
|
|
"margin": 0.5,
|
|
"_ts": pytest.approx(strategy.contexts[0]["routing_history"][0]["_ts"]),
|
|
}
|
|
],
|
|
"prev_assistant_text": "Previous assistant answer.",
|
|
"prev_assistant_usage": {"output_tokens": 321},
|
|
"history_user_texts": ["First user question.", "Second user question."],
|
|
"flags_text_override": "Continue from the previous answer.",
|
|
}
|
|
]
|
|
|
|
|
|
def test_v4_request_contains_current_history_assistant_and_route_context() -> None:
|
|
class FakeInferenceRequest:
|
|
def __init__(self, **kwargs) -> None:
|
|
self.__dict__.update(kwargs)
|
|
|
|
strategy = V4Phase3Strategy(require_router_runtime=False)
|
|
strategy._request_type = FakeInferenceRequest
|
|
|
|
request = strategy._build_request(
|
|
"Current user question.",
|
|
[
|
|
{
|
|
"text": "Previous user question.",
|
|
"final_route_class": "R2",
|
|
"difficulty_score": 2.0,
|
|
"margin": 0.7,
|
|
}
|
|
],
|
|
prev_assistant_text="Previous assistant answer.",
|
|
prev_assistant_usage={"output_tokens": 456},
|
|
history_user_texts=["Earlier user question.", "Previous user question."],
|
|
flags_text_override="Current user question.",
|
|
)
|
|
|
|
assert request.current_user_text == "Current user question."
|
|
assert request.history_user_texts == ["Earlier user question.", "Previous user question."]
|
|
assert request.prev_assistant_text == "Previous assistant answer."
|
|
assert request.prev_assistant_usage == {"output_tokens": 456}
|
|
assert request.prev_route_decisions[0].route_class == "R2"
|
|
assert request.prev_route_decisions[0].difficulty == 2.0
|
|
assert request.prev_route_decisions[0].margin == 0.7
|
|
assert request.flags_text_override == "Current user question."
|
|
assert request.context_metadata["history_user_turn_count"] == 2
|
|
assert request.context_metadata["has_prev_assistant"] is True
|
|
assert request.context_metadata["context_tokens_est"] > 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_image_input_routes_directly_to_vision_model_without_prompt_injection(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
monkeypatch.setattr(
|
|
squilla_router_step,
|
|
"_get_strategy",
|
|
lambda _config: pytest.fail("image routing should not invoke text strategy"),
|
|
)
|
|
ctx = make_context(
|
|
"What is in this screenshot?",
|
|
attachments=[{"type": "image", "mime_type": "image/png"}],
|
|
)
|
|
|
|
routed = await apply_squilla_router(ctx)
|
|
|
|
assert routed.model == "moonshotai/kimi-k2.6"
|
|
assert routed.metadata["routed_tier"] == "image_model"
|
|
assert routed.metadata["routed_model"] == "moonshotai/kimi-k2.6"
|
|
assert routed.metadata["routing_applied"] is True
|
|
assert routed.metadata["routing_confidence"] == 1.0
|
|
assert routed.metadata["routing_source"] == "image_route"
|
|
assert routed.metadata["route_max_history_turns"] == 1
|
|
assert routed.metadata["thinking_requested"] is True
|
|
assert routed.metadata["thinking_level"] == "medium"
|
|
assert "[RESPONSE_POLICY:" not in routed.message
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_image_route_uses_first_configured_image_tier_without_random_choice(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
monkeypatch.setattr(
|
|
squilla_router_step,
|
|
"_get_strategy",
|
|
lambda _config: pytest.fail("image routing should not invoke text strategy"),
|
|
)
|
|
monkeypatch.setattr(
|
|
"random.choice",
|
|
lambda _items: pytest.fail("image route should be deterministic"),
|
|
)
|
|
ctx = make_context(
|
|
"Describe this screenshot.",
|
|
attachments=[{"type": "image/png", "data": "abc"}],
|
|
)
|
|
ctx.config.squilla_router.tiers = {
|
|
"vision_primary": {
|
|
"model": "vision/primary",
|
|
"supports_image": True,
|
|
"thinking_level": "low",
|
|
},
|
|
"vision_backup": {
|
|
"model": "vision/backup",
|
|
"supports_image": True,
|
|
"thinking_level": "high",
|
|
},
|
|
"c1": {"model": "text/model"},
|
|
}
|
|
|
|
routed = await apply_squilla_router(ctx)
|
|
|
|
assert routed.metadata["routed_tier"] == "vision_primary"
|
|
assert routed.metadata["routed_model"] == "vision/primary"
|
|
assert routed.model == "vision/primary"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_image_route_records_tier_provider_for_cross_provider_execution(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
monkeypatch.setattr(
|
|
squilla_router_step,
|
|
"_get_strategy",
|
|
lambda _config: pytest.fail("image routing should not invoke text strategy"),
|
|
)
|
|
ctx = make_context(
|
|
"What is in this screenshot?",
|
|
attachments=[{"type": "image", "mime_type": "image/png"}],
|
|
)
|
|
ctx.config.llm.provider = "anthropic"
|
|
ctx.config.squilla_router.cross_provider_tiers = True
|
|
ctx.config.squilla_router.tiers = {
|
|
"image_model": {
|
|
"model": "vision/model-1",
|
|
"provider": "openai",
|
|
"supports_image": True,
|
|
"image_only": True,
|
|
},
|
|
"c1": {"model": "text/model-1"},
|
|
}
|
|
|
|
routed = await apply_squilla_router(ctx)
|
|
|
|
assert routed.metadata["routing_source"] == "image_route"
|
|
assert routed.metadata["routing_applied"] is True
|
|
assert routed.metadata.get("routed_provider") == "openai"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_image_route_flags_tier_provider_mismatch_when_cross_provider_disabled(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
monkeypatch.setattr(
|
|
squilla_router_step,
|
|
"_get_strategy",
|
|
lambda _config: pytest.fail("image routing should not invoke text strategy"),
|
|
)
|
|
ctx = make_context(
|
|
"Describe this screenshot.",
|
|
attachments=[{"type": "image", "mime_type": "image/png"}],
|
|
)
|
|
ctx.config.llm.provider = "anthropic"
|
|
ctx.config.squilla_router.cross_provider_tiers = False
|
|
ctx.config.squilla_router.tiers = {
|
|
"image_model": {
|
|
"model": "vision/model-1",
|
|
"provider": "openai",
|
|
"supports_image": True,
|
|
"image_only": True,
|
|
},
|
|
"c1": {"model": "text/model-1"},
|
|
}
|
|
|
|
routed = await apply_squilla_router(ctx)
|
|
|
|
assert routed.metadata["routing_source"] == "image_route"
|
|
assert routed.metadata.get("router_tier_provider_mismatch") == "openai"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_gate_needs_image_routes_followup_to_vision_model(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
monkeypatch.setattr(
|
|
squilla_router_step,
|
|
"_get_strategy",
|
|
lambda _config: pytest.fail("gate image routing should not invoke text strategy"),
|
|
)
|
|
ctx = make_context("Continue from that image.")
|
|
ctx.metadata["router_history_has_recent_image"] = True
|
|
ctx.metadata["router_history_image_turn_count"] = 2
|
|
ctx.metadata["router_turns_since_last_image"] = 1
|
|
ctx.metadata["router_vision_followup_gate_decision"] = "needs_image"
|
|
ctx.metadata["router_vision_followup_needs_image"] = True
|
|
|
|
routed = await apply_squilla_router(ctx)
|
|
|
|
assert routed.model == "moonshotai/kimi-k2.6"
|
|
assert routed.metadata["routed_tier"] == "image_model"
|
|
assert routed.metadata["routing_source"] == "image_route"
|
|
assert routed.metadata["image_route_reason"] == "gate_history"
|
|
assert routed.metadata["route_max_history_turns"] == 8
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sticky_without_gate_no_longer_routes_to_vision_model(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
strategy = fake_strategy(
|
|
monkeypatch,
|
|
"c2",
|
|
0.77,
|
|
{"route_class": "R2", "thinking_mode": "T1", "prompt_policy": "P1"},
|
|
)
|
|
ctx = make_context("Write a Python script.")
|
|
ctx.metadata["router_history_has_recent_image"] = True
|
|
ctx.metadata["router_history_image_turn_count"] = 1
|
|
ctx.metadata["router_vision_sticky_remaining"] = 3
|
|
ctx.metadata["router_vision_followup_gate_decision"] = "text_only"
|
|
ctx.metadata["router_vision_followup_needs_image"] = False
|
|
|
|
routed = await apply_squilla_router(ctx)
|
|
|
|
assert strategy.calls == 1
|
|
assert routed.metadata["routing_source"] != "image_route"
|
|
assert routed.metadata.get("image_route_reason") is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_recent_historical_image_without_sticky_uses_text_router(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
strategy = fake_strategy(
|
|
monkeypatch,
|
|
"c0",
|
|
0.88,
|
|
{"route_class": "R0", "thinking_mode": "T1", "prompt_policy": "P0"},
|
|
)
|
|
ctx = make_context("Help me write a Python script.")
|
|
ctx.metadata["router_history_has_recent_image"] = True
|
|
ctx.metadata["router_history_image_turn_count"] = 1
|
|
|
|
routed = await apply_squilla_router(ctx)
|
|
|
|
assert strategy.calls == 1
|
|
assert routed.metadata["routing_source"] == "v4_phase3"
|
|
assert routed.metadata["routed_tier"] == "c0"
|
|
assert "image_route_reason" not in routed.metadata
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_image_attachment_without_image_tier_fails_locally(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
monkeypatch.setattr(
|
|
squilla_router_step,
|
|
"_get_strategy",
|
|
lambda _config: pytest.fail(
|
|
"image routing without image tier should not invoke text strategy"
|
|
),
|
|
)
|
|
ctx = make_context(
|
|
"What is in this screenshot?",
|
|
attachments=[{"type": "image", "mime_type": "image/png"}],
|
|
)
|
|
ctx.config.squilla_router.tiers["image_model"]["supports_image"] = False
|
|
|
|
with pytest.raises(
|
|
RuntimeError,
|
|
match="No image-capable SquillaRouter tier is configured",
|
|
):
|
|
await apply_squilla_router(ctx)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_caption_less_image_attachment_still_routes_to_vision_tier(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
monkeypatch.setattr(
|
|
squilla_router_step,
|
|
"_get_strategy",
|
|
lambda _config: pytest.fail("image routing should not invoke text strategy"),
|
|
)
|
|
for caption in ("", " \n\t "):
|
|
ctx = make_context(
|
|
caption,
|
|
attachments=[{"type": "image", "mime_type": "image/png"}],
|
|
)
|
|
|
|
routed = await apply_squilla_router(ctx)
|
|
|
|
assert routed.metadata["routing_source"] == "image_route"
|
|
assert routed.metadata["routed_tier"] == "image_model"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_caption_less_image_attachment_without_image_tier_fails_locally(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
monkeypatch.setattr(
|
|
squilla_router_step,
|
|
"_get_strategy",
|
|
lambda _config: pytest.fail(
|
|
"image routing without image tier should not invoke text strategy"
|
|
),
|
|
)
|
|
ctx = make_context(
|
|
"",
|
|
attachments=[{"type": "image", "mime_type": "image/png"}],
|
|
)
|
|
ctx.config.squilla_router.tiers["image_model"]["supports_image"] = False
|
|
|
|
with pytest.raises(
|
|
RuntimeError,
|
|
match="No image-capable SquillaRouter tier is configured",
|
|
):
|
|
await apply_squilla_router(ctx)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_non_image_attachment_does_not_force_vision_route(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
strategy = fake_strategy(
|
|
monkeypatch,
|
|
"c1",
|
|
0.91,
|
|
{
|
|
"route_class": "R1",
|
|
"thinking_mode": "T1",
|
|
"prompt_policy": "P0",
|
|
},
|
|
)
|
|
ctx = make_context(
|
|
"Summarize the attached PDF text.",
|
|
attachments=[{"type": "application/pdf", "mime_type": "application/pdf"}],
|
|
)
|
|
|
|
routed = await apply_squilla_router(ctx)
|
|
|
|
assert strategy.calls == 1
|
|
assert routed.metadata["routing_source"] == "v4_phase3"
|
|
assert routed.metadata["routed_tier"] == "c1"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_observe_rollout_records_decisions_without_applying_model_or_prompt(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
fake_strategy(
|
|
monkeypatch,
|
|
"c2",
|
|
0.93,
|
|
{
|
|
"route_class": "R2",
|
|
"thinking_mode": "T2",
|
|
"prompt_policy": "P2",
|
|
"prompt_hint": "Use extra care.",
|
|
},
|
|
)
|
|
ctx = make_context("Analyze this code path.", rollout_phase="observe")
|
|
baseline_model = ctx.model
|
|
|
|
routed = await apply_squilla_router(ctx)
|
|
|
|
assert routed.model == baseline_model
|
|
assert routed.metadata["routed_tier"] == "c2"
|
|
assert routed.metadata["routed_model"] == "z-ai/glm-5.2"
|
|
assert routed.metadata["routing_applied"] is False
|
|
assert routed.metadata["thinking_mode"] == "T2"
|
|
assert routed.metadata["thinking_level"] == "medium"
|
|
assert "[RESPONSE_POLICY:" not in routed.message
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_repeated_message_across_sessions_is_classified_each_time(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
strategy = fake_strategy(
|
|
monkeypatch,
|
|
"c1",
|
|
0.91,
|
|
{
|
|
"route_class": "R1",
|
|
"thinking_mode": "T1",
|
|
"prompt_policy": "P0",
|
|
},
|
|
)
|
|
|
|
first = await apply_squilla_router(make_context("Repeat this.", session_key="session-a"))
|
|
second = await apply_squilla_router(make_context("Repeat this.", session_key="session-b"))
|
|
|
|
assert first.metadata["routing_source"] == "v4_phase3"
|
|
assert second.metadata["routing_source"] == "v4_phase3"
|
|
assert strategy.calls == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_required_router_runtime_failure_falls_back_to_heuristic_tiering(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
import opensquilla.squilla_router.v4_phase3 as v4_phase3
|
|
|
|
monkeypatch.setattr(v4_phase3, "V4Phase3Strategy", ExplodingV4Strategy)
|
|
ctx = make_context("Explain the setup steps.")
|
|
ctx.config.squilla_router.require_router_runtime = True
|
|
|
|
routed = await apply_squilla_router(ctx)
|
|
|
|
# Short plain text lands in the heuristic c0 band; the decision is
|
|
# honestly attributed to the fallback classifier, not the ML runtime.
|
|
assert routed.metadata["routing_source"] == "heuristic"
|
|
assert routed.metadata["routed_tier"] == "c0"
|
|
assert routed.metadata["routing_confidence"] == pytest.approx(0.55)
|
|
assert routed.metadata["routing_extra"]["heuristic_band"] == "short_plain"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_router_runtime_failure_emits_one_operator_warning(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
caplog: pytest.LogCaptureFixture,
|
|
) -> None:
|
|
import opensquilla.squilla_router.v4_phase3 as v4_phase3
|
|
|
|
monkeypatch.setattr(v4_phase3, "V4Phase3Strategy", ExplodingV4Strategy)
|
|
caplog.set_level(logging.WARNING)
|
|
|
|
first = make_context("Explain the setup steps.")
|
|
first.config.squilla_router.require_router_runtime = True
|
|
await apply_squilla_router(first)
|
|
squilla_router_step._strategy = None
|
|
squilla_router_step._strategy_key = None
|
|
second = make_context("Explain the setup steps again.")
|
|
second.config.squilla_router.require_router_runtime = True
|
|
await apply_squilla_router(second)
|
|
|
|
messages = [
|
|
record.getMessage()
|
|
for record in caplog.records
|
|
if "Microsoft Visual C++ Redistributable 2015-2022 x64" in record.getMessage()
|
|
]
|
|
assert len(messages) == 1
|
|
assert "safe router fallback" in messages[0]
|
|
assert "https://aka.ms/vs/17/release/vc_redist.x64.exe" in messages[0]
|
|
assert "After installing, reopen PowerShell and restart OpenSquilla" in messages[0]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_router_runtime_failure_emits_macos_libomp_guidance(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
caplog: pytest.LogCaptureFixture,
|
|
) -> None:
|
|
import opensquilla.squilla_router.v4_phase3 as v4_phase3
|
|
|
|
monkeypatch.setattr(v4_phase3, "V4Phase3Strategy", MacLibompExplodingV4Strategy)
|
|
caplog.set_level(logging.WARNING)
|
|
|
|
ctx = make_context("Explain the setup steps.")
|
|
ctx.config.squilla_router.require_router_runtime = True
|
|
routed = await apply_squilla_router(ctx)
|
|
|
|
assert routed.metadata["routing_source"] == "heuristic"
|
|
messages = [
|
|
record.getMessage()
|
|
for record in caplog.records
|
|
if "brew install libomp" in record.getMessage()
|
|
]
|
|
assert len(messages) == 1
|
|
assert "default-tier routing" in messages[0]
|
|
assert "opensquilla gateway restart" in messages[0]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_runtime_router_short_chinese_prompt_injects_localized_p0_hint() -> None:
|
|
require_runtime_router()
|
|
ctx = make_context("直接总结这句话。")
|
|
|
|
routed = await apply_squilla_router(ctx)
|
|
|
|
assert routed.metadata["routing_source"] == "v4_phase3"
|
|
assert routed.metadata["routed_tier"] == "c0"
|
|
assert routed.model == "deepseek/deepseek-v4-flash"
|
|
assert routed.metadata["thinking_mode"] == "T0"
|
|
assert routed.metadata.get("thinking_requested") is None
|
|
assert routed.metadata["prompt_policy"] == "P0"
|
|
assert "[RESPONSE_POLICY: 直接作答，缩短思考长度，避免无关展开。]" in routed.message
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_runtime_router_complex_request_applies_deep_thinking_without_p2_prompt() -> None:
|
|
require_runtime_router()
|
|
ctx = make_context("Plan a risky multi-step database migration with rollback and verification.")
|
|
|
|
routed = await apply_squilla_router(ctx)
|
|
|
|
assert routed.metadata["routing_source"] == "v4_phase3"
|
|
assert routed.metadata["routed_tier"] == "c3"
|
|
assert routed.model == "anthropic/claude-opus-4.8"
|
|
assert routed.metadata["thinking_mode"] == "T3"
|
|
assert routed.metadata["thinking_requested"] is True
|
|
assert routed.metadata["thinking_level"] == "high"
|
|
assert routed.metadata["prompt_policy"] == "P2"
|
|
assert routed.metadata["routing_extra"]["prompt_hint"] == (
|
|
"Analyze thoroughly, cover key constraints, avoid omissions."
|
|
)
|
|
assert "[RESPONSE_POLICY:" not in routed.message
|