776 lines
24 KiB
Python
776 lines
24 KiB
Python
"""Contracts for Agent Hermes automatic route selection."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
|
|
SOURCE = (
|
|
Path(__file__).parents[2]
|
|
/ "services/hermes/plugins/auto-router/__init__.py"
|
|
)
|
|
SPEC = importlib.util.spec_from_file_location("hermes_auto_router", SOURCE)
|
|
assert SPEC and SPEC.loader
|
|
router = importlib.util.module_from_spec(SPEC)
|
|
sys.modules[SPEC.name] = router
|
|
SPEC.loader.exec_module(router)
|
|
|
|
|
|
def _status() -> dict:
|
|
return {
|
|
"providers": {
|
|
"openai-codex": {"connected": True},
|
|
"anthropic": {"connected": True},
|
|
},
|
|
"routes": {
|
|
"codex-low": [
|
|
"openai-codex/gpt-5.6-luna",
|
|
"anthropic/claude-haiku-4-5-20251001",
|
|
],
|
|
"codex-medium": [
|
|
"openai-codex/gpt-5.6-terra",
|
|
"anthropic/claude-sonnet-5",
|
|
],
|
|
"claude-medium": [
|
|
"anthropic/claude-sonnet-5",
|
|
"openai-codex/gpt-5.6-terra",
|
|
],
|
|
"claude-xhigh": [
|
|
"anthropic/claude-opus-5",
|
|
"openai-codex/gpt-5.6-sol",
|
|
],
|
|
},
|
|
}
|
|
|
|
|
|
def test_heuristics_keep_simple_questions_cheap_and_risky_work_capped():
|
|
simple = router.heuristic_decision("Who is the current provider?")
|
|
risky = router.heuristic_decision("Migrate production Vault credentials safely")
|
|
|
|
assert (simple.shape, simple.effort, simple.provider) == (
|
|
"question",
|
|
"low",
|
|
"codex",
|
|
)
|
|
assert (risky.shape, risky.effort, risky.provider) == (
|
|
"review",
|
|
"xhigh",
|
|
"claude",
|
|
)
|
|
|
|
|
|
def test_jetson_selects_provider_while_deterministic_policy_preserves_work_shape(
|
|
monkeypatch,
|
|
):
|
|
monkeypatch.setattr(
|
|
router,
|
|
"jetson_decision",
|
|
lambda text: router.Decision(
|
|
"question", "high", "claude", "jetson", "test", 42
|
|
),
|
|
)
|
|
|
|
decision = router.classify_task("Implement and test the new API handler")
|
|
|
|
assert decision.shape == "implementation"
|
|
assert decision.provider == "claude"
|
|
assert decision.effort == "high"
|
|
assert decision.classifier == "jetson"
|
|
|
|
|
|
def test_route_uses_managed_models_and_connected_provider_fallback():
|
|
status = _status()
|
|
decision = router.Decision(
|
|
"question", "low", "codex", "heuristic", "short question"
|
|
)
|
|
plan = router.select_route(status, decision)
|
|
assert plan["profile"] == "codex-low"
|
|
assert plan["model"] == "gpt-5.6-luna"
|
|
|
|
status["providers"]["openai-codex"]["connected"] = False
|
|
status["routes"]["claude-low"] = [
|
|
"anthropic/claude-haiku-4-5-20251001",
|
|
"openai-codex/gpt-5.6-luna",
|
|
]
|
|
fallback = router.select_route(status, decision)
|
|
assert fallback["profile"] == "claude-low"
|
|
assert fallback["provider"] == "anthropic"
|
|
|
|
|
|
def test_route_circuit_breaker_skips_recently_failed_provider():
|
|
status = _status()
|
|
status["routes"]["codex-medium"] = [
|
|
"openai-codex/gpt-5.6-terra",
|
|
"anthropic/claude-sonnet-5",
|
|
]
|
|
decision = router.Decision(
|
|
"question", "medium", "claude", "jetson", "local vote"
|
|
)
|
|
policy = {
|
|
"mode": "auto",
|
|
"provider_cooldowns": {
|
|
"anthropic": {"until_epoch": router.time.time() + 300}
|
|
},
|
|
}
|
|
|
|
plan = router.select_route(status, decision, policy=policy)
|
|
|
|
assert plan["profile"] == "codex-medium"
|
|
assert plan["provider"] == "openai-codex"
|
|
|
|
|
|
def test_local_classifier_accepts_only_bounded_route_decisions():
|
|
assert router._validated_local_route("?", "?", "?", 1) is None
|
|
decision = router._validated_local_route("A", "H", "D", 1)
|
|
assert decision is not None
|
|
assert (decision.shape, decision.provider, decision.effort, decision.priority) == (
|
|
"question",
|
|
"claude",
|
|
"high",
|
|
"deep",
|
|
)
|
|
assert router._validated_local_route("C", "M", "?", 1) is None
|
|
|
|
|
|
def test_structured_vote_requires_a_complete_bounded_json_object():
|
|
assert router._parse_route_vote(
|
|
'{"provider":"C","effort":"M","priority":"F"}'
|
|
) == ("C", "M", "F")
|
|
assert router._parse_route_vote('"A"') is None
|
|
assert router._parse_route_vote(
|
|
'{"provider":"codex","effort":"M","priority":"F"}'
|
|
) is None
|
|
assert router._parse_route_vote(
|
|
'{"provider":"A","effort":"H"}'
|
|
) is None
|
|
|
|
|
|
def test_jetson_requests_one_structured_provider_effort_priority_vote(monkeypatch):
|
|
calls = []
|
|
|
|
def structured(text, timeout):
|
|
calls.append((text, timeout))
|
|
return (("A", "H", "D"), 12)
|
|
|
|
monkeypatch.setattr(router, "_jetson_route", structured)
|
|
|
|
decision = router.jetson_decision("Review the architecture")
|
|
|
|
assert (
|
|
decision.provider,
|
|
decision.effort,
|
|
decision.priority,
|
|
decision.latency_ms,
|
|
) == (
|
|
"claude",
|
|
"high",
|
|
"deep",
|
|
12,
|
|
)
|
|
assert calls == [("Review the architecture", 2.5)]
|
|
|
|
|
|
def test_every_auto_classification_consults_jetson_and_keeps_safety_floors(monkeypatch):
|
|
calls = []
|
|
|
|
def classify(text):
|
|
calls.append(text)
|
|
return router.Decision("question", "low", "codex", "jetson", "test", 5)
|
|
|
|
monkeypatch.setattr(router, "jetson_decision", classify)
|
|
|
|
simple = router.classify_task("Who is the provider?")
|
|
risky = router.classify_task("Migrate production Vault credentials")
|
|
|
|
assert len(calls) == 2
|
|
assert (simple.effort, simple.provider) == ("low", "codex")
|
|
assert (risky.shape, risky.effort, risky.provider) == (
|
|
"review",
|
|
"xhigh",
|
|
"claude",
|
|
)
|
|
|
|
|
|
def test_trivial_prompt_respects_balanced_jetson_effort(monkeypatch):
|
|
calls = []
|
|
|
|
def classify(text):
|
|
calls.append(text)
|
|
return router.Decision("question", "medium", "codex", "jetson", "test", 5)
|
|
|
|
monkeypatch.setattr(router, "jetson_decision", classify)
|
|
|
|
decision = router.classify_task(
|
|
"Reply with exactly ROUTE_SMOKE_OK. Do not call tools."
|
|
)
|
|
|
|
assert calls == ["Reply with exactly ROUTE_SMOKE_OK. Do not call tools."]
|
|
assert (decision.effort, decision.provider, decision.classifier) == (
|
|
"medium",
|
|
"codex",
|
|
"jetson",
|
|
)
|
|
|
|
|
|
def test_semantic_speed_priority_reduces_speculative_depth_but_not_safety(monkeypatch):
|
|
monkeypatch.setattr(
|
|
router,
|
|
"jetson_decision",
|
|
lambda text: router.Decision(
|
|
"question", "high", "codex", "jetson", "test", 5, "fast"
|
|
),
|
|
)
|
|
|
|
simple = router.classify_task("Give me a concise status summary.")
|
|
risky = router.classify_task("Quickly migrate production Vault credentials.")
|
|
|
|
assert simple.effort == "medium"
|
|
assert risky.effort == "xhigh"
|
|
|
|
|
|
def test_ui_priority_changes_quality_posture_without_crossing_safety_floor(monkeypatch):
|
|
monkeypatch.setattr(
|
|
router,
|
|
"jetson_decision",
|
|
lambda text: router.Decision(
|
|
"question", "low", "codex", "jetson", "test", 5, "balanced"
|
|
),
|
|
)
|
|
|
|
maximum = router.classify_task(
|
|
"Give me the current status.", priority_override="maximum"
|
|
)
|
|
fast_risky = router.classify_task(
|
|
"Delete production Vault credentials.", priority_override="fast"
|
|
)
|
|
|
|
assert (maximum.priority, maximum.effort, maximum.classifier) == (
|
|
"maximum",
|
|
"high",
|
|
"ui-jetson",
|
|
)
|
|
assert (fast_risky.priority, fast_risky.effort, fast_risky.provider) == (
|
|
"fast",
|
|
"xhigh",
|
|
"claude",
|
|
)
|
|
|
|
|
|
def test_service_fallback_postures_favor_chat_speed_and_agent_quality(monkeypatch):
|
|
monkeypatch.setattr(router, "jetson_decision", lambda text: None)
|
|
|
|
monkeypatch.setattr(router, "ROUTER_PROFILE", "chat")
|
|
monkeypatch.setattr(router, "CHAT_MODE", True)
|
|
chat = router.classify_task("What time is dinner?")
|
|
|
|
monkeypatch.setattr(router, "ROUTER_PROFILE", "triage")
|
|
monkeypatch.setattr(router, "CHAT_MODE", False)
|
|
triage = router.classify_task("Summarize the failed health check.")
|
|
|
|
monkeypatch.setattr(router, "ROUTER_PROFILE", "agent")
|
|
agent = router.classify_task("Explain this helper function.")
|
|
|
|
assert (chat.priority, chat.effort, chat.provider) == ("fast", "low", "local")
|
|
assert (triage.priority, triage.effort) == ("deep", "medium")
|
|
assert (agent.priority, agent.effort) == ("maximum", "high")
|
|
|
|
|
|
def test_trivial_prompt_can_still_escalate_on_strong_jetson_signal(monkeypatch):
|
|
monkeypatch.setattr(
|
|
router,
|
|
"jetson_decision",
|
|
lambda text: router.Decision(
|
|
"question", "high", "claude", "jetson", "test", 5
|
|
),
|
|
)
|
|
|
|
decision = router.classify_task("Check this.")
|
|
|
|
assert (decision.effort, decision.provider) == ("high", "claude")
|
|
|
|
|
|
def test_architecture_and_review_fail_upward_to_claude(monkeypatch):
|
|
monkeypatch.setattr(
|
|
router,
|
|
"jetson_decision",
|
|
lambda text: router.Decision(
|
|
"question", "low", "codex", "jetson", "test", 5
|
|
),
|
|
)
|
|
|
|
architecture = router.classify_task("Design the service architecture")
|
|
review = router.classify_task("Review this change for regressions")
|
|
|
|
assert (architecture.provider, architecture.effort) == ("claude", "medium")
|
|
assert (review.provider, review.effort) == ("claude", "medium")
|
|
|
|
|
|
def test_referential_outstanding_work_never_uses_low_route(monkeypatch):
|
|
monkeypatch.setattr(
|
|
router,
|
|
"jetson_decision",
|
|
lambda text: router.Decision(
|
|
"question", "low", "codex", "jetson", "test", 10
|
|
),
|
|
)
|
|
|
|
decision = router.classify_task(
|
|
"Look, loop through all of the still outstanding work that you identified. "
|
|
"Do it to the best of your abilities."
|
|
)
|
|
|
|
assert (decision.shape, decision.effort, decision.provider) == (
|
|
"implementation",
|
|
"high",
|
|
"codex",
|
|
)
|
|
|
|
|
|
def test_referential_followup_uses_recent_context_for_risk_floor(monkeypatch):
|
|
calls = []
|
|
|
|
def classify(text):
|
|
calls.append(text)
|
|
return router.Decision("question", "low", "codex", "jetson", "test", 5)
|
|
|
|
monkeypatch.setattr(router, "jetson_decision", classify)
|
|
history = [
|
|
{"role": "user", "content": "Is the work complete?"},
|
|
{
|
|
"role": "assistant",
|
|
"content": [
|
|
{
|
|
"type": "text",
|
|
"text": (
|
|
"Still outstanding: production deployment retry, provider "
|
|
"switching, runtime experiments, and a full repository test pass."
|
|
),
|
|
}
|
|
],
|
|
},
|
|
]
|
|
|
|
decision = router.classify_task(
|
|
"Loop through all outstanding work and finish it.", history
|
|
)
|
|
|
|
assert (decision.shape, decision.effort, decision.provider) == (
|
|
"review",
|
|
"xhigh",
|
|
"claude",
|
|
)
|
|
assert decision.classifier == "jetson-context"
|
|
assert "recent assistant context" in decision.reason
|
|
assert len(calls) == 1
|
|
|
|
|
|
def test_internal_prompt_contains_objective_tools_and_results():
|
|
text = router._internal_task_text(
|
|
"Finish the production deployment safely",
|
|
[
|
|
{
|
|
"role": "assistant",
|
|
"content": "I will inspect the failed rollout.",
|
|
"tool_calls": [
|
|
{
|
|
"function": {
|
|
"name": "terminal",
|
|
"arguments": '{"command":"kubectl get pods"}',
|
|
}
|
|
}
|
|
],
|
|
},
|
|
{"role": "tool", "content": "deployment is degraded"},
|
|
],
|
|
)
|
|
|
|
assert "Finish the production deployment safely" in text
|
|
assert "planned tool terminal" in text
|
|
assert "deployment is degraded" in text
|
|
|
|
|
|
def test_every_internal_auto_prompt_is_reclassified_and_applied(monkeypatch):
|
|
calls = []
|
|
monkeypatch.setattr(router, "_current_policy", lambda: {"mode": "auto"})
|
|
monkeypatch.setattr(router, "_load_json", lambda path: _status())
|
|
monkeypatch.setattr(
|
|
router,
|
|
"classify_task",
|
|
lambda text: calls.append(text)
|
|
or router.Decision(
|
|
"question", "medium", "claude", "jetson", "test", 7
|
|
),
|
|
)
|
|
applied = []
|
|
recorded = []
|
|
monkeypatch.setattr(
|
|
router, "_apply_route", lambda ctx, agent, plan: applied.append(plan)
|
|
)
|
|
monkeypatch.setattr(
|
|
router,
|
|
"_record_internal_plan",
|
|
lambda policy, plan, count: recorded.append((plan, count)),
|
|
)
|
|
|
|
class Agent:
|
|
provider = "openai-codex"
|
|
model = "gpt-5.6-luna"
|
|
reasoning_config = {"effort": "low"}
|
|
|
|
def _emit_status(self, message):
|
|
self.message = message
|
|
|
|
agent = Agent()
|
|
router._pre_internal_route(
|
|
object(),
|
|
agent=agent,
|
|
user_message="Continue",
|
|
conversation_history=[
|
|
{"role": "tool", "content": "The architecture review found a risk."}
|
|
],
|
|
api_call_count=3,
|
|
)
|
|
|
|
assert len(calls) == 1
|
|
assert applied[0]["profile"] == "claude-medium"
|
|
assert applied[0]["classifier"] == "jetson-internal"
|
|
assert recorded[0][1] == 3
|
|
assert agent.message.startswith("AUTO internal #3")
|
|
|
|
|
|
def test_manual_route_audits_internal_prompt_without_overriding_user_choice(monkeypatch):
|
|
monkeypatch.setattr(
|
|
router,
|
|
"_current_policy",
|
|
lambda: {
|
|
"mode": "manual",
|
|
"manual": {"provider": "claude", "effort": "medium", "model": ""},
|
|
},
|
|
)
|
|
monkeypatch.setattr(router, "_load_json", lambda path: _status())
|
|
calls = []
|
|
monkeypatch.setattr(
|
|
router,
|
|
"classify_task",
|
|
lambda text: calls.append(text)
|
|
or router.Decision("implementation", "xhigh", "codex", "jetson", "audit", 5),
|
|
)
|
|
plans = []
|
|
monkeypatch.setattr(router, "_apply_route", lambda ctx, agent, plan: plans.append(plan))
|
|
monkeypatch.setattr(router, "_record_internal_plan", lambda *args: None)
|
|
|
|
router._pre_internal_route(
|
|
object(),
|
|
agent=object(),
|
|
user_message="Continue",
|
|
conversation_history=[{"role": "tool", "content": "done"}],
|
|
api_call_count=2,
|
|
)
|
|
assert len(calls) == 1
|
|
assert plans[0]["profile"] == "claude-medium"
|
|
assert plans[0]["classifier"] == "manual-jetson-internal"
|
|
|
|
|
|
def test_every_native_subagent_is_classified_and_routed_independently(monkeypatch):
|
|
calls = []
|
|
monkeypatch.setattr(router, "_current_policy", lambda: {"mode": "auto"})
|
|
monkeypatch.setattr(router, "_load_json", lambda path: _status())
|
|
monkeypatch.setattr(
|
|
router,
|
|
"classify_task",
|
|
lambda text: calls.append(text)
|
|
or router.Decision(
|
|
"implementation", "medium", "codex", "jetson", "test", 9
|
|
),
|
|
)
|
|
applied = []
|
|
recorded = []
|
|
monkeypatch.setattr(
|
|
router, "_apply_route", lambda ctx, agent, plan: applied.append((agent, plan))
|
|
)
|
|
monkeypatch.setattr(
|
|
router,
|
|
"_record_subagent_plan",
|
|
lambda policy, plan, goal, index: recorded.append((plan, goal, index)),
|
|
)
|
|
|
|
class Parent:
|
|
def _emit_status(self, message):
|
|
self.message = message
|
|
|
|
child = object()
|
|
parent = Parent()
|
|
router._pre_subagent_route(
|
|
object(),
|
|
agent=child,
|
|
parent_agent=parent,
|
|
goal="Implement the bounded parser fix",
|
|
context="Run the focused tests.",
|
|
task_index=2,
|
|
)
|
|
|
|
assert len(calls) == 1
|
|
assert "Run the focused tests" in calls[0]
|
|
assert applied[0][0] is child
|
|
assert applied[0][1]["profile"] == "codex-medium"
|
|
assert applied[0][1]["classifier"] == "jetson-subagent"
|
|
assert recorded[0][1:] == ("Implement the bounded parser fix", 2)
|
|
assert parent.message.startswith("AUTO child #3")
|
|
|
|
|
|
def test_manual_route_audits_and_applies_override_to_native_subagent(monkeypatch):
|
|
monkeypatch.setattr(
|
|
router,
|
|
"_current_policy",
|
|
lambda: {
|
|
"mode": "manual",
|
|
"manual": {"provider": "claude", "effort": "medium", "model": ""},
|
|
},
|
|
)
|
|
monkeypatch.setattr(router, "_load_json", lambda path: _status())
|
|
calls = []
|
|
monkeypatch.setattr(
|
|
router,
|
|
"classify_task",
|
|
lambda text: calls.append(text)
|
|
or router.Decision("implementation", "high", "codex", "jetson", "audit", 5),
|
|
)
|
|
plans = []
|
|
monkeypatch.setattr(router, "_apply_route", lambda ctx, agent, plan: plans.append(plan))
|
|
monkeypatch.setattr(router, "_record_subagent_plan", lambda *args: None)
|
|
|
|
router._pre_subagent_route(
|
|
object(), agent=object(), parent_agent=object(), goal="Review the diff"
|
|
)
|
|
assert len(calls) == 1
|
|
assert plans[0]["profile"] == "claude-medium"
|
|
assert plans[0]["classifier"] == "manual-jetson-subagent"
|
|
|
|
|
|
def test_manual_policy_is_reapplied_on_every_non_command_turn(monkeypatch):
|
|
monkeypatch.setattr(
|
|
router,
|
|
"_current_policy",
|
|
lambda: {
|
|
"mode": "manual",
|
|
"manual": {"provider": "claude", "effort": "medium", "model": ""},
|
|
},
|
|
)
|
|
monkeypatch.setattr(router, "_load_json", lambda path: _status())
|
|
monkeypatch.setattr(
|
|
router,
|
|
"classify_task",
|
|
lambda text, history=None: router.Decision(
|
|
"implementation", "high", "codex", "jetson", "audit", 5
|
|
),
|
|
)
|
|
plans = []
|
|
monkeypatch.setattr(router, "_apply_route", lambda ctx, agent, plan: plans.append(plan))
|
|
monkeypatch.setattr(router, "_record_plan", lambda policy, plan: None)
|
|
|
|
class Agent:
|
|
def _emit_status(self, message):
|
|
self.message = message
|
|
|
|
agent = Agent()
|
|
router._pre_turn_route(object(), agent=agent, user_message="Continue the task")
|
|
|
|
assert plans[0]["profile"] == "claude-medium"
|
|
assert plans[0]["model"] == "claude-sonnet-5"
|
|
assert agent.message.startswith("MANUAL target")
|
|
assert plans[0]["classifier"] == "manual-jetson"
|
|
|
|
|
|
def test_webui_exact_model_and_effort_remain_authoritative_after_jetson_audit(
|
|
monkeypatch,
|
|
):
|
|
monkeypatch.setattr(router, "CHAT_MODE", True)
|
|
monkeypatch.setattr(router, "PROVIDERS", ("codex", "claude", "local"))
|
|
monkeypatch.setattr(router, "_current_policy", lambda: {"mode": "auto"})
|
|
monkeypatch.setattr(router, "_load_json", lambda path: {})
|
|
audits = []
|
|
|
|
def classify(text, history=None, priority_override=""):
|
|
audits.append((text, priority_override))
|
|
return router.Decision(
|
|
"question", "low", "local", "jetson", "audit", 8, "fast"
|
|
)
|
|
|
|
monkeypatch.setattr(router, "classify_task", classify)
|
|
plans = []
|
|
monkeypatch.setattr(router, "_apply_route", lambda ctx, agent, plan: plans.append(plan))
|
|
monkeypatch.setattr(router, "_record_plan", lambda policy, plan: None)
|
|
|
|
class Agent:
|
|
provider = "openai-codex"
|
|
model = "gpt-5.6-sol"
|
|
_hermes_routing_priority = "deep"
|
|
_hermes_explicit_model_pick = True
|
|
_hermes_explicit_reasoning_effort = "xhigh"
|
|
|
|
def _emit_status(self, message):
|
|
self.message = message
|
|
|
|
agent = Agent()
|
|
router._pre_turn_route(
|
|
object(), agent=agent, user_message="Review this answer carefully."
|
|
)
|
|
|
|
assert audits == [("Review this answer carefully.", "deep")]
|
|
assert plans[0]["provider"] == "atlas-codex"
|
|
assert plans[0]["model"] == "gpt-5.6-sol"
|
|
assert plans[0]["effort"] == "xhigh"
|
|
assert plans[0]["classifier"] == "manual-ui-jetson"
|
|
assert agent.message.startswith("MANUAL target")
|
|
|
|
|
|
def test_chat_natural_language_override_is_one_turn_and_keeps_jetson_audit(
|
|
monkeypatch,
|
|
):
|
|
monkeypatch.setattr(router, "CHAT_MODE", True)
|
|
monkeypatch.setattr(router, "PROVIDERS", ("codex", "claude", "local"))
|
|
monkeypatch.setattr(router, "_current_policy", lambda: {"mode": "auto"})
|
|
monkeypatch.setattr(router, "_load_json", lambda path: {})
|
|
calls = []
|
|
monkeypatch.setattr(
|
|
router,
|
|
"classify_task",
|
|
lambda text, history=None: calls.append(text)
|
|
or router.Decision("question", "low", "local", "jetson", "audit", 8),
|
|
)
|
|
plans = []
|
|
monkeypatch.setattr(router, "_apply_route", lambda ctx, agent, plan: plans.append(plan))
|
|
monkeypatch.setattr(router, "_record_plan", lambda policy, plan: None)
|
|
|
|
class Agent:
|
|
def _emit_status(self, message):
|
|
self.message = message
|
|
|
|
agent = Agent()
|
|
router._pre_turn_route(
|
|
object(), agent=agent, user_message="Use Claude at xhigh for this answer."
|
|
)
|
|
|
|
assert calls == ["Use Claude at xhigh for this answer."]
|
|
assert plans[0]["provider"] == "anthropic"
|
|
assert plans[0]["model"] == "claude-opus-5"
|
|
assert plans[0]["effort"] == "xhigh"
|
|
assert plans[0]["classifier"] == "explicit-jetson"
|
|
assert agent.message.startswith("USER target")
|
|
|
|
|
|
def test_chat_text_overrides_do_not_steal_image_provider_instructions(monkeypatch):
|
|
monkeypatch.setattr(router, "CHAT_MODE", True)
|
|
|
|
assert router._explicit_text_override("Use local image generation for this photo") is None
|
|
assert router._explicit_text_override("Generate this image with OpenAI") is None
|
|
assert router._explicit_text_override("Answer locally with the Qwen model") == (
|
|
"local",
|
|
"",
|
|
)
|
|
assert router._explicit_text_override("Ask Codex with high reasoning") == (
|
|
"codex",
|
|
"high",
|
|
)
|
|
|
|
|
|
def test_post_turn_records_and_announces_capacity_fallback(monkeypatch):
|
|
policy = {
|
|
"mode": "auto",
|
|
"last_decision": {
|
|
"provider": "anthropic",
|
|
"model": "claude-sonnet-5",
|
|
"effort": "medium",
|
|
"classifier": "jetson",
|
|
},
|
|
}
|
|
written = []
|
|
monkeypatch.setattr(router, "_current_policy", lambda: policy)
|
|
monkeypatch.setattr(router, "_write_policy", lambda value: written.append(value))
|
|
|
|
class Agent:
|
|
provider = "openai-codex"
|
|
model = "gpt-5.6-terra"
|
|
|
|
def _emit_status(self, message):
|
|
self.message = message
|
|
|
|
agent = Agent()
|
|
cli = type("CLI", (), {"agent": agent})()
|
|
manager = type("Manager", (), {"_cli_ref": cli})()
|
|
ctx = type("Context", (), {"_manager": manager})()
|
|
|
|
router._post_turn_route(ctx, model="gpt-5.6-terra")
|
|
|
|
outcome = written[-1]["last_decision"]
|
|
assert outcome["fallback_used"] is True
|
|
assert outcome["actual_provider"] == "openai-codex"
|
|
assert outcome["actual_model"] == "gpt-5.6-terra"
|
|
assert written[-1]["provider_cooldowns"]["anthropic"]["until_epoch"] > router.time.time()
|
|
assert agent.message.startswith("FALLBACK USED")
|
|
|
|
|
|
def test_post_turn_rewarms_classifier_after_local_chat(monkeypatch):
|
|
policy = {
|
|
"mode": "auto",
|
|
"last_decision": {
|
|
"provider": "custom",
|
|
"model": "qwen2.5:14b-instruct-q4_0",
|
|
"effort": "low",
|
|
"classifier": "jetson",
|
|
},
|
|
}
|
|
warmed = []
|
|
monkeypatch.setattr(router, "_current_policy", lambda: policy)
|
|
monkeypatch.setattr(router, "_write_policy", lambda value: None)
|
|
monkeypatch.setattr(
|
|
router,
|
|
"_rewarm_classifier_after_local",
|
|
lambda provider, model: warmed.append((provider, model)),
|
|
)
|
|
|
|
class Agent:
|
|
provider = "custom"
|
|
model = "qwen2.5:14b-instruct-q4_0"
|
|
|
|
def _emit_status(self, message):
|
|
self.message = message
|
|
|
|
agent = Agent()
|
|
cli = type("CLI", (), {"agent": agent})()
|
|
manager = type("Manager", (), {"_cli_ref": cli})()
|
|
ctx = type("Context", (), {"_manager": manager})()
|
|
|
|
router._post_turn_route(ctx, model="qwen2.5:14b-instruct-q4_0")
|
|
|
|
assert warmed == [("custom", "qwen2.5:14b-instruct-q4_0")]
|
|
assert agent.message.startswith("ROUTE USED")
|
|
|
|
|
|
def test_status_distinguishes_requested_route_from_actual_outcome(monkeypatch):
|
|
monkeypatch.setattr(
|
|
router,
|
|
"_current_policy",
|
|
lambda: {
|
|
"mode": "auto",
|
|
"last_decision": {
|
|
"provider": "anthropic",
|
|
"model": "claude-sonnet-5",
|
|
"effort": "medium",
|
|
"classifier": "jetson",
|
|
"actual_provider": "openai-codex",
|
|
"actual_model": "gpt-5.6-terra",
|
|
"fallback_used": True,
|
|
},
|
|
},
|
|
)
|
|
manager = type("Manager", (), {"_cli_ref": None})()
|
|
ctx = type("Context", (), {"_manager": manager})()
|
|
|
|
status = router._status_text(ctx)
|
|
|
|
assert "Last requested route: anthropic/claude-sonnet-5" in status
|
|
assert "Last actual outcome: fallback: openai-codex/gpt-5.6-terra" in status
|