fix(hermes): harden delegation router safety

cascadeflow/integrations/hermes/__init__.py

@@ -26,4 +26,3 @@
     "extract_cascadeflow_skill_metadata",
     "profile_from_skill_metadata",
 ]
-

cascadeflow/integrations/hermes/classifier.py

@@ -151,4 +151,3 @@ def _detect_complexity(self, text: str) -> tuple[str, float]:
         else:
             complexity_value = str(complexity)
         return complexity_value, max(0.0, min(1.0, float(confidence or 0.0)))
-

cascadeflow/integrations/hermes/router.py

@@ -66,9 +66,10 @@ def _route_delegation(
             )
 
         classification = self.classifier.classify(request)
-        route_key = self._select_route_key(classification.domain, classification.complexity)
-        profile = self.config.routes.get(route_key) if route_key else None
-        if profile is None and classification.domain in HIGH_STAKES_DOMAINS:
+        if (
+            classification.domain in HIGH_STAKES_DOMAINS
+            and classification.domain not in self.config.routes
+        ):
             return HermesDelegationDecision(
                 action="inherit",
                 domain=classification.domain,
@@ -77,8 +78,14 @@ def _route_delegation(
                 confidence=classification.confidence,
                 reason="high_stakes_domain_unconfigured",
                 source="classifier",
-                metadata={"route_key": route_key},
+                metadata={
+                    "route_key": classification.domain,
+                    "loaded_skills": list(request.loaded_skills),
+                },
             )
+
+        route_key = self._select_route_key(classification.domain, classification.complexity)
+        profile = self.config.routes.get(route_key) if route_key else None
         if profile is None:
             return HermesDelegationDecision(
                 action="inherit",
@@ -88,7 +95,10 @@ def _route_delegation(
                 confidence=classification.confidence,
                 reason="no_matching_route",
                 source="classifier",
-                metadata={"route_key": route_key},
+                metadata={
+                    "route_key": route_key,
+                    "loaded_skills": list(request.loaded_skills),
+                },
             )
 
         return self._decision_from_profile(
@@ -139,9 +149,7 @@ def _decision_from_profile(
 
         provider = profile.provider if self.config.route_provider_model else None
         model = profile.model if self.config.route_provider_model else None
-        reasoning_effort = (
-            profile.reasoning_effort if self.config.route_reasoning_effort else None
-        )
+        reasoning_effort = profile.reasoning_effort if self.config.route_reasoning_effort else None
         return HermesDelegationDecision(
             action=action,
             provider=provider,

cascadeflow/integrations/hermes/types.py

@@ -30,11 +30,7 @@ def _clean_tuple(values: Any) -> tuple[str, ...]:
     if isinstance(values, str):
         values = (values,)
     try:
-        return tuple(
-            item
-            for item in (_clean_text(v, max_length=120) for v in values)
-            if item
-        )
+        return tuple(item for item in (_clean_text(v, max_length=120) for v in values) if item)
     except TypeError:
         return ()
 
@@ -78,8 +74,6 @@ def text_for_classification(self) -> str:
         parts = [self.goal]
         if self.context:
             parts.append(self.context)
-        if self.loaded_skills:
-            parts.append(" ".join(self.loaded_skills))
         return "\n".join(part for part in parts if part)
 
 

examples/integrations/hermes_delegation_router.py

@@ -94,4 +94,3 @@ def route_before_delegate_task(
     )
     print("\nSkill-routed subagent decision:")
     print(skill_decision)
-

tests/integrations/test_hermes_classifier.py

@@ -0,0 +1,127 @@
+"""Coverage tests for HermesTaskClassifier — exercise real domain/complexity paths."""
+
+import pytest
+
+from cascadeflow.integrations.hermes import (
+    HermesDelegationRequest,
+    HermesTaskClassifier,
+)
+from cascadeflow.integrations.hermes.classifier import (
+    DOMAIN_KEYWORDS,
+    HIGH_STAKES_DOMAINS,
+    TOOLSET_DOMAIN_HINTS,
+)
+
+
+def _req(goal, **kw):
+    return HermesDelegationRequest(goal=goal, **kw)
+
+
+@pytest.fixture
+def classifier():
+    return HermesTaskClassifier()
+
+
+@pytest.mark.parametrize(
+    ("domain", "goal"),
+    [
+        ("debugging", "Investigate the failing test traceback and find the root cause"),
+        ("code", "Implement a TypeScript API client with unit tests for the schema"),
+        ("research", "Research and summarize recent web sources on the market"),
+        ("data", "Run SQL against the postgres warehouse and build a dashboard with metrics"),
+        ("creative", "Write headline copy and brand script for the design"),
+        ("ops", "Deploy the kubernetes workflow and review CI logs for the incident"),
+        ("legal", "Review the contract for privacy compliance and terms"),
+        ("medical", "Summarize the patient diagnosis and treatment options"),
+        ("finance", "Reconcile the invoice and calculate the tax on revenue"),
+    ],
+)
+def test_each_domain_keyword_set_classifies(classifier, domain, goal):
+    cls = classifier.classify(_req(goal))
+    assert cls.domain == domain
+    assert cls.confidence > 0.5
+
+
+def test_no_keywords_falls_back_to_general(classifier):
+    cls = classifier.classify(_req("hello there my friend"))
+    # generic chitchat with simple/trivial complexity → mapped to "simple"
+    assert cls.domain in {"general", "simple"}
+
+
+def test_simple_complexity_remaps_general_to_simple(classifier):
+    cls = classifier.classify(_req("hi"))
+    assert cls.domain == "simple"
+    assert cls.reason == "simple_complexity"
+    assert cls.topic is None
+
+
+def test_topic_set_for_non_simple_domain(classifier):
+    cls = classifier.classify(_req("Refactor this python schema to add unit tests"))
+    assert cls.domain == "code"
+    assert cls.topic == "code"
+
+
+def test_toolset_hints_boost_domain(classifier):
+    # No code keywords in goal, but git toolset hints to code
+    cls = classifier.classify(_req("Please assist with this task", toolsets=("git",)))
+    assert cls.domain == "code"
+
+
+def test_browser_toolset_hints_to_research(classifier):
+    cls = classifier.classify(_req("Look this up for me", toolsets=("browser",)))
+    assert cls.domain == "research"
+
+
+def test_high_stakes_domains_are_detected(classifier):
+    for domain in HIGH_STAKES_DOMAINS:
+        keyword = next(iter(DOMAIN_KEYWORDS[domain]))
+        cls = classifier.classify(_req(f"Help with {keyword} matters"))
+        assert cls.domain == domain
+        assert cls.domain in HIGH_STAKES_DOMAINS
+
+
+def test_keyword_hits_increase_confidence(classifier):
+    one_hit = classifier.classify(_req("review the contract"))
+    many_hits = classifier.classify(
+        _req("review the contract for legal compliance and privacy terms under the law")
+    )
+    assert many_hits.confidence >= one_hit.confidence
+    assert many_hits.confidence <= 0.95  # capped
+
+
+def test_confidence_clamped_to_unit_interval(classifier):
+    # Stack every keyword for one domain; confidence must still <= 1.0
+    keywords = " ".join(DOMAIN_KEYWORDS["code"])
+    cls = classifier.classify(_req(keywords))
+    assert 0.0 <= cls.confidence <= 1.0
+
+
+def test_text_for_classification_includes_context(classifier):
+    # Goal is empty of signal; debugging keywords live only in context
+    cls = classifier.classify(
+        _req(
+            "vague goal",
+            context="Find the bug and debug the regression to identify root cause",
+        )
+    )
+    assert cls.domain == "debugging"
+
+
+def test_classifier_handles_failing_complexity_detector():
+    class Boom:
+        def detect(self, text):
+            raise RuntimeError("boom")
+
+    classifier = HermesTaskClassifier(complexity_detector=Boom())
+    cls = classifier.classify(_req("Implement the python schema"))
+    # falls back to moderate complexity rather than crashing
+    assert cls.complexity == "moderate"
+    assert cls.domain == "code"
+
+
+def test_toolset_domain_hints_table_is_consistent():
+    # Every hinted domain must exist in DOMAIN_KEYWORDS (otherwise routing breaks)
+    for hinted_domain in TOOLSET_DOMAIN_HINTS.values():
+        assert (
+            hinted_domain in DOMAIN_KEYWORDS
+        ), f"toolset hint points to unknown domain {hinted_domain!r}"

tests/integrations/test_hermes_router.py

@@ -234,6 +234,64 @@ def test_high_stakes_domain_without_matching_route_inherits():
     assert decision.domain == "medical"
 
 
+def test_high_stakes_domain_does_not_fall_back_to_complexity_route():
+    router = _router(
+        {
+            "enabled": True,
+            "mode": "route",
+            "routes": {
+                "simple": {
+                    "provider": "openai",
+                    "model": "gpt-4.1-mini",
+                    "reasoning_effort": "low",
+                },
+                "general": {
+                    "provider": "openai",
+                    "model": "gpt-4.1-mini",
+                    "reasoning_effort": "medium",
+                },
+            },
+        },
+        domain="legal",
+        complexity="simple",
+        confidence=0.86,
+    )
+
+    decision = router.route_delegation(_request(goal="Review this contract clause"))
+
+    assert decision.action == "inherit"
+    assert decision.reason == "high_stakes_domain_unconfigured"
+    assert decision.domain == "legal"
+    assert decision.model is None
+
+
+def test_parent_loaded_skills_do_not_override_task_domain():
+    router = HermesDelegationRouter.from_dict(
+        {
+            "enabled": True,
+            "mode": "route",
+            "routes": {
+                "code": {
+                    "provider": "nous",
+                    "model": "nous/hermes-4.1",
+                    "reasoning_effort": "high",
+                }
+            },
+        }
+    )
+    request = HermesDelegationRequest(
+        goal="Review this indemnity clause for legal risk",
+        loaded_skills=("python", "debugging"),
+    )
+
+    decision = router.route_delegation(request)
+
+    assert decision.action == "inherit"
+    assert decision.reason == "high_stakes_domain_unconfigured"
+    assert decision.domain == "legal"
+    assert decision.metadata["loaded_skills"] == ["python", "debugging"]
+
+
 def test_confidence_below_threshold_inherits_but_keeps_audit_recommendation():
     router = _router(
         {