feat(gateway): add smart_model_routing for auto-complexity classification

gateway/run.py

@@ -2513,6 +2513,39 @@ def _recover_telegram_topic_thread_id(
                 return None
         return None
 
+    def _apply_smart_routing(self, message_text: str, model: str, runtime_kwargs: dict, smart_cfg: dict, user_config: dict) -> tuple[str, dict]:
+        """Classify message complexity and route to cheap_model if simple.
+
+        Why: Cuts cost by sending trivial gateway messages (status/show/check)
+        to a cheap model before the agent runs, without changing behavior for
+        complex requests or manual /model overrides.
+        What: Returns (cheap_model, swapped_kwargs) when the message is short and
+        free of complexity keywords; otherwise returns (model, kwargs) unchanged.
+        Test: Assert a short "status" message returns cheap_model; a message
+        containing "implement" or longer than max_simple_chars returns model unchanged.
+        """
+        cheap_model = smart_cfg.get("cheap_model", "")
+        if not cheap_model or cheap_model == model:
+            return model, runtime_kwargs
+
+        text = (message_text or "").strip()
+        max_chars = int(smart_cfg.get("max_simple_chars", 200))
+        max_words = int(smart_cfg.get("max_simple_words", 40))
+        complexity_kw = [k.lower() for k in smart_cfg.get("complexity_keywords", [])]
+        text_lower = text.lower()
+
+        # Complexity signals override simple signals
+        if any(kw in text_lower for kw in complexity_kw):
+            return model, runtime_kwargs
+
+        word_count = len(text.split())
+        if len(text) <= max_chars and word_count <= max_words:
+            cheap_kwargs = dict(runtime_kwargs)
+            cheap_kwargs["model"] = cheap_model
+            return cheap_model, cheap_kwargs
+
+        return model, runtime_kwargs
+
     def _resolve_session_agent_runtime(
         self,
         *,
@@ -17402,6 +17435,15 @@ def run_sync():
                     "run_agent resolved: model=%s provider=%s session=%s",
                     model, runtime_kwargs.get("provider"), session_key or "",
                 )
+
+                # smart_model_routing: auto-classify complexity before agent dispatch
+                if not self._session_model_overrides.get(session_key):
+                    _smart_cfg = (user_config or {}).get("smart_model_routing") or {}
+                    _platform = getattr(source, "platform", None)
+                    if _smart_cfg.get("enabled") and str(_platform) != "Platform.LOCAL":
+                        model, runtime_kwargs = self._apply_smart_routing(
+                            message, model, runtime_kwargs, _smart_cfg, user_config
+                        )
             except Exception as exc:
                 return {
                     "final_response": f"⚠️ Provider authentication failed: {exc}",

hermes_cli/config.py

@@ -2284,7 +2284,15 @@ def _ensure_hermes_home_managed(home: Path):
 
 
     # Config schema version - bump this when adding new required fields
-    "_config_version": 25,
+    "smart_model_routing": {
+        "enabled": False,
+        "cheap_model": "deepseek/deepseek-v4-flash",
+        "max_simple_chars": 200,
+        "max_simple_words": 40,
+        "complexity_keywords": ["implement", "debug", "refactor", "diagnose", "migrate", "architect", "explain", "why does", "how does", "broken", "failing"],
+        "simple_keywords": ["status", "show", "check", "list", "restart", "what is", "ping", "health"],
+    },
+    "_config_version": 26,
 }
 
 # =============================================================================

tests/gateway/test_smart_model_routing.py

@@ -0,0 +1,100 @@
+"""Unit tests for gateway-level smart_model_routing complexity classifier.
+
+Why: smart_model_routing swaps simple gateway messages onto a cheap model
+before agent dispatch; these tests lock in the classification contract and the
+run_sync gate guards (manual /model override wins, disabled config is a no-op)
+so cost-routing never silently misfires on complex or operator-controlled work.
+What: Exercises GatewayRunner._apply_smart_routing directly for routing
+decisions, plus replicates the run_sync gate predicate for override/disabled.
+Test: Run `pytest tests/gateway/test_smart_model_routing.py`.
+"""
+
+from types import SimpleNamespace
+
+from gateway.run import GatewayRunner
+
+
+DEFAULT_MODEL = "openai/gpt-oss-120b"
+CHEAP_MODEL = "deepseek/deepseek-v4-flash"
+
+SMART_CFG = {
+    "enabled": True,
+    "cheap_model": CHEAP_MODEL,
+    "max_simple_chars": 200,
+    "max_simple_words": 40,
+    "complexity_keywords": [
+        "implement", "debug", "refactor", "diagnose", "migrate",
+        "architect", "explain", "why does", "how does", "broken", "failing",
+    ],
+    "simple_keywords": ["status", "show", "check", "list", "restart", "what is", "ping", "health"],
+}
+
+
+def _route(message_text, model=DEFAULT_MODEL, runtime_kwargs=None, smart_cfg=None):
+    """Call _apply_smart_routing without building a full GatewayRunner.
+
+    Why: The method reads no instance state, so an unbound call keeps the test
+    fast and free of gateway construction side effects.
+    """
+    runtime_kwargs = runtime_kwargs if runtime_kwargs is not None else {"provider": "openrouter"}
+    smart_cfg = smart_cfg if smart_cfg is not None else SMART_CFG
+    dummy = SimpleNamespace()
+    return GatewayRunner._apply_smart_routing(
+        dummy, message_text, model, runtime_kwargs, smart_cfg, {}
+    )
+
+
+def test_short_simple_message_routes_to_cheap_model():
+    """Why: A short status check is the canonical cost-saving target."""
+    model, kwargs = _route("status")
+    assert model == CHEAP_MODEL
+    assert kwargs["model"] == CHEAP_MODEL
+    # provider/credentials must be preserved on the swapped kwargs
+    assert kwargs["provider"] == "openrouter"
+
+
+def test_complexity_keyword_keeps_default_model():
+    """Why: Complexity signals must override shortness to protect hard work."""
+    model, kwargs = _route("implement smart routing")
+    assert model == DEFAULT_MODEL
+    assert "model" not in kwargs  # original kwargs untouched
+
+
+def test_over_max_chars_keeps_default_model():
+    """Why: Long messages are presumed complex regardless of keywords."""
+    long_text = "a " * 150  # 300 chars, 150 words, no complexity keyword
+    model, _ = _route(long_text.strip())
+    assert model == DEFAULT_MODEL
+
+
+def test_session_override_skips_routing():
+    """Why: Manual /model override always wins; the run_sync gate must skip routing."""
+    session_overrides = {"sess-1": {"model": "anthropic/claude"}}
+    session_key = "sess-1"
+    # Replicates the run_sync gate predicate verbatim.
+    should_route = not session_overrides.get(session_key)
+    assert should_route is False
+    # When skipped, model is whatever resolve returned (unchanged).
+    model = DEFAULT_MODEL
+    if should_route and SMART_CFG.get("enabled"):
+        model, _ = _route("status")
+    assert model == DEFAULT_MODEL
+
+
+def test_disabled_config_skips_routing():
+    """Why: enabled=False must be a no-op even for trivially simple messages."""
+    disabled_cfg = dict(SMART_CFG, enabled=False)
+    session_overrides = {}
+    session_key = "sess-2"
+    model = DEFAULT_MODEL
+    if not session_overrides.get(session_key) and disabled_cfg.get("enabled"):
+        model, _ = _route("status", smart_cfg=disabled_cfg)
+    assert model == DEFAULT_MODEL
+
+
+def test_same_cheap_and_default_is_noop():
+    """Why: Guard against pointless swap when cheap_model == active model."""
+    cfg = dict(SMART_CFG, cheap_model=DEFAULT_MODEL)
+    model, kwargs = _route("status", smart_cfg=cfg)
+    assert model == DEFAULT_MODEL
+    assert "model" not in kwargs

tests/honcho_plugin/test_session.py

@@ -1,5 +1,7 @@
 """Tests for plugins/memory/honcho/session.py — HonchoSession and helpers."""
 
+import time
+
 from datetime import datetime
 from types import SimpleNamespace
 from unittest.mock import MagicMock
@@ -1538,8 +1540,27 @@ def _make_provider(cfg_extra=None):
             return provider, mock_manager, cfg
 
     def _await_thread(self, provider):
-        if provider._prefetch_thread:
-            provider._prefetch_thread.join(timeout=3.0)
+        """Block until the in-flight prefetch/prewarm thread has fully finished.
+
+        The earlier version did a single ``join(timeout=3.0)`` and then
+        proceeded regardless of whether the thread had actually finished. On a
+        loaded CI runner (6 parallel test slices), the background dialectic
+        thread's completion can slip past that 3s window, so the join times out
+        silently and the test reads ``_prefetch_result`` before the worker wrote
+        it — a flaky ``session-start prewarm must land`` failure. We instead join
+        in a loop up to a generous ceiling and assert the thread is dead, so a
+        genuine hang surfaces as a clear, non-flaky failure instead of a race.
+        """
+        thread = provider._prefetch_thread
+        if thread is None:
+            return
+        deadline = time.monotonic() + 30.0
+        while thread.is_alive() and time.monotonic() < deadline:
+            thread.join(timeout=1.0)
+        assert not thread.is_alive(), (
+            "prefetch/prewarm thread did not finish within 30s — "
+            "this is a real hang, not a timing flake"
+        )
 
     def test_full_multi_turn_session(self):
         """Walks init → turns 1..8 → session end. Asserts at every step that