diff --git a/cascadeflow/integrations/hermes/__init__.py b/cascadeflow/integrations/hermes/__init__.py index d7175a2b..ee0c409d 100644 --- a/cascadeflow/integrations/hermes/__init__.py +++ b/cascadeflow/integrations/hermes/__init__.py @@ -26,4 +26,3 @@ "extract_cascadeflow_skill_metadata", "profile_from_skill_metadata", ] - diff --git a/cascadeflow/integrations/hermes/classifier.py b/cascadeflow/integrations/hermes/classifier.py index 8c7b87a3..eb5f15ac 100644 --- a/cascadeflow/integrations/hermes/classifier.py +++ b/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))) - diff --git a/cascadeflow/integrations/hermes/router.py b/cascadeflow/integrations/hermes/router.py index b16f2f2c..d78aed03 100644 --- a/cascadeflow/integrations/hermes/router.py +++ b/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, diff --git a/cascadeflow/integrations/hermes/types.py b/cascadeflow/integrations/hermes/types.py index 7b25c3b1..ba99d2d1 100644 --- a/cascadeflow/integrations/hermes/types.py +++ b/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) diff --git a/examples/integrations/hermes_delegation_router.py b/examples/integrations/hermes_delegation_router.py index 2450d262..414c3c8f 100644 --- a/examples/integrations/hermes_delegation_router.py +++ b/examples/integrations/hermes_delegation_router.py @@ -94,4 +94,3 @@ def route_before_delegate_task( ) print("\nSkill-routed subagent decision:") print(skill_decision) - diff --git a/tests/integrations/test_hermes_classifier.py b/tests/integrations/test_hermes_classifier.py new file mode 100644 index 00000000..618c44d7 --- /dev/null +++ b/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}" diff --git a/tests/integrations/test_hermes_router.py b/tests/integrations/test_hermes_router.py index c6e871ea..80a849d1 100644 --- a/tests/integrations/test_hermes_router.py +++ b/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( { diff --git a/tests/integrations/test_hermes_router_edges.py b/tests/integrations/test_hermes_router_edges.py new file mode 100644 index 00000000..aa000e9a --- /dev/null +++ b/tests/integrations/test_hermes_router_edges.py @@ -0,0 +1,242 @@ +"""Edge-case / fuzz sweep for HermesDelegationRouter. + +Targets: +- route_key fallback chain (domain → debugging→code → complexity → simple/hard → general) +- no_matching_route path +- malformed / partial configs via from_dict +- skill_metadata variations +- request crash paths +""" + +import pytest + +from cascadeflow.integrations.hermes import ( + HermesDelegationRequest, + HermesDelegationRouter, + HermesRoutingConfig, +) + + +def _req(**kw): + return HermesDelegationRequest(goal=kw.pop("goal", "Implement the python API"), **kw) + + +# --- route_key fallback chain --- + + +def test_debugging_domain_falls_back_to_code_route(): + router = HermesDelegationRouter.from_dict( + { + "enabled": True, + "mode": "route", + "routes": {"code": {"provider": "nous", "model": "nous/hermes-4.1"}}, + } + ) + decision = router.route_delegation( + _req(goal="Investigate the failing test traceback and find the root cause") + ) + assert decision.action == "route" + assert decision.provider == "nous" + assert decision.domain == "debugging" + + +def test_simple_complexity_falls_back_to_simple_route(): + router = HermesDelegationRouter.from_dict( + { + "enabled": True, + "mode": "route", + "min_confidence": 0.0, + "routes": {"simple": {"provider": "nous", "model": "nous/hermes-flash"}}, + } + ) + # Trivial goal → general/simple complexity → simple route + decision = router.route_delegation(_req(goal="hi")) + assert decision.action == "route" + assert decision.model == "nous/hermes-flash" + + +def test_general_route_used_when_nothing_else_matches(): + router = HermesDelegationRouter.from_dict( + { + "enabled": True, + "mode": "route", + "min_confidence": 0.0, + "routes": {"general": {"provider": "nous", "model": "nous/hermes-general"}}, + } + ) + # Creative goal with no creative route → falls back to general + decision = router.route_delegation(_req(goal="Write a creative brand headline")) + assert decision.action == "route" + assert decision.model == "nous/hermes-general" + + +def test_no_matching_route_returns_inherit_with_no_matching_route_reason(): + router = HermesDelegationRouter.from_dict( + { + "enabled": True, + "mode": "route", + "routes": {"code": {"provider": "nous", "model": "nous/hermes-4.1"}}, + } + ) + decision = router.route_delegation(_req(goal="Write a creative brand headline")) + assert decision.action == "inherit" + assert decision.reason == "no_matching_route" + assert decision.domain == "creative" + + +# --- malformed config --- + + +def test_from_dict_handles_none(): + router = HermesDelegationRouter.from_dict(None) + decision = router.route_delegation(_req()) + # default config has enabled=False → inherit + assert decision.action == "inherit" + + +def test_from_dict_handles_empty(): + router = HermesDelegationRouter.from_dict({}) + decision = router.route_delegation(_req()) + assert decision.action == "inherit" + + +def test_from_dict_ignores_unknown_keys(): + router = HermesDelegationRouter.from_dict( + { + "enabled": True, + "mode": "route", + "this_key_does_not_exist": "garbage", + "routes": {"code": {"provider": "nous", "model": "nous/hermes-4.1"}}, + } + ) + decision = router.route_delegation(_req()) + assert decision.action == "route" + + +def test_from_dict_handles_malformed_route_profile(): + # route value is not a dict → config should tolerate or skip + router = HermesDelegationRouter.from_dict( + { + "enabled": True, + "mode": "route", + "routes": {"code": "this is not a dict"}, + } + ) + decision = router.route_delegation(_req()) + # Either gracefully skipped or surfaced via "router_error" fallback + assert decision.action == "inherit" + + +# --- skill_metadata variations --- + + +def test_skill_metadata_route_overrides_classifier(): + router = HermesDelegationRouter.from_dict({"enabled": True, "mode": "route", "routes": {}}) + decision = router.route_delegation( + _req( + goal="Some legal contract review", # would otherwise be inherit/legal + skill_metadata={ + "cascadeflow": { + "provider": "nous", + "model": "nous/hermes-explicit", + "domain": "compliance", + } + }, + ) + ) + # explicit skill metadata wins + assert decision.source == "skill_metadata" + assert decision.model == "nous/hermes-explicit" + + +def test_empty_skill_metadata_falls_through_to_classifier(): + router = HermesDelegationRouter.from_dict( + { + "enabled": True, + "mode": "route", + "routes": {"code": {"provider": "nous", "model": "nous/hermes-4.1"}}, + } + ) + decision = router.route_delegation(_req(goal="Implement the python API", skill_metadata={})) + assert decision.source == "classifier" + assert decision.action == "route" + + +def test_skill_metadata_with_unknown_fields_is_tolerated(): + router = HermesDelegationRouter.from_dict({"enabled": True, "mode": "route", "routes": {}}) + decision = router.route_delegation( + _req( + goal="any", + skill_metadata={ + "cascadeflow": { + "provider": "nous", + "model": "nous/hermes-4.1", + "this_field_is_unknown": "garbage", + "nested": {"more": "garbage"}, + } + }, + ) + ) + assert decision.source == "skill_metadata" + + +# --- request crash paths --- + + +def test_router_returns_fallback_decision_on_unexpected_exception(): + router = HermesDelegationRouter(config=HermesRoutingConfig(enabled=True)) + # Inject a request that lies about its toolsets type (not str-coerced) + request = _req(goal="any") + + # Force classifier to blow up by stubbing + class BoomClassifier: + def classify(self, request): + raise RuntimeError("kaboom") + + router.classifier = BoomClassifier() + decision = router.route_delegation(request) + assert decision.action == "inherit" + assert decision.source == "fallback" + assert decision.reason == "router_error" + assert "RuntimeError" in decision.metadata.get("error", "") + + +# --- observe mode --- + + +def test_observe_mode_returns_inherit_but_keeps_route_metadata(): + router = HermesDelegationRouter.from_dict( + { + "enabled": True, + "mode": "observe", + "min_confidence": 0.0, + "routes": {"code": {"provider": "nous", "model": "nous/hermes-4.1"}}, + } + ) + decision = router.route_delegation(_req(goal="Implement the python API")) + assert decision.action == "inherit" + # In observe mode, recommended provider/model should be None (not applied), + # but the metadata about the would-be-route is preserved on the decision + assert decision.domain == "code" + + +# --- per-route confidence --- + + +def test_per_route_confidence_can_exceed_classifier_confidence(): + router = HermesDelegationRouter.from_dict( + { + "enabled": True, + "mode": "route", + "min_confidence": 0.0, + "routes": { + "code": { + "provider": "nous", + "model": "nous/hermes-4.1", + "confidence": 0.99, + } + }, + } + ) + decision = router.route_delegation(_req(goal="Implement the python API")) + assert decision.confidence >= 0.99 diff --git a/tests/integrations/test_hermes_skill_metadata.py b/tests/integrations/test_hermes_skill_metadata.py index 9defc084..0f16fc01 100644 --- a/tests/integrations/test_hermes_skill_metadata.py +++ b/tests/integrations/test_hermes_skill_metadata.py @@ -47,4 +47,3 @@ def test_extracts_nested_hermes_cascadeflow_metadata(): def test_missing_skill_metadata_returns_no_profile(): assert extract_cascadeflow_skill_metadata({"name": "plain-skill"}) == {} assert profile_from_skill_metadata({"name": "plain-skill"}) is None - diff --git a/tests/integrations/test_hermes_types.py b/tests/integrations/test_hermes_types.py new file mode 100644 index 00000000..6af74ddf --- /dev/null +++ b/tests/integrations/test_hermes_types.py @@ -0,0 +1,166 @@ +"""Edge-case coverage for HermesDelegationRequest / HermesDelegationDecision sanitization.""" + +import pytest + +from cascadeflow.integrations.hermes import ( + HermesDelegationDecision, + HermesDelegationRequest, +) +from cascadeflow.integrations.hermes.types import ( + VALID_ACTIONS, + VALID_REASONING_EFFORTS, + _clean_text, + _clean_tuple, +) + + +# --- _clean_text --- + + +def test_clean_text_strips_newlines_and_control_chars(): + assert _clean_text("hello\nworld\r\nfoo") == "hello world foo" + + +def test_clean_text_truncates_with_ellipsis(): + long = "a" * 600 + out = _clean_text(long, max_length=10) + assert out == "aaaaaaa..." + assert len(out) == 10 + + +def test_clean_text_handles_none(): + assert _clean_text(None) == "" + + +def test_clean_text_drops_non_printable(): + assert _clean_text("hi\x00\x07there") == "hithere" + + +# --- _clean_tuple --- + + +def test_clean_tuple_handles_string_input(): + assert _clean_tuple("solo") == ("solo",) + + +def test_clean_tuple_handles_none(): + assert _clean_tuple(None) == () + + +def test_clean_tuple_skips_empty_entries(): + assert _clean_tuple(["a", "", None, " ", "b"]) == ("a", "b") + + +def test_clean_tuple_handles_non_iterable(): + assert _clean_tuple(42) == () + + +# --- HermesDelegationRequest --- + + +def test_request_sanitizes_goal_and_truncates(): + long_goal = "x" * 3000 + req = HermesDelegationRequest(goal=long_goal) + assert len(req.goal) == 2000 + assert req.goal.endswith("...") + + +def test_request_defaults_role_to_leaf_when_empty(): + req = HermesDelegationRequest(goal="hi", role="") + assert req.role == "leaf" + + +def test_request_normalizes_string_toolset_to_tuple(): + req = HermesDelegationRequest(goal="hi", toolsets="terminal") + assert req.toolsets == ("terminal",) + + +def test_text_for_classification_with_empty_extras(): + req = HermesDelegationRequest(goal="just a goal") + assert req.text_for_classification() == "just a goal" + + +def test_text_for_classification_excludes_loaded_skills(): + # loaded_skills are parent context for audit, NOT a classification signal — + # see test_parent_loaded_skills_do_not_override_task_domain in test_hermes_router.py + req = HermesDelegationRequest(goal="g", loaded_skills=("python", "git")) + assert req.text_for_classification() == "g" + + +# --- HermesDelegationDecision --- + + +def test_decision_normalizes_invalid_action_to_inherit(): + d = HermesDelegationDecision(action="DESTROY_ALL_HUMANS") + assert d.action == "inherit" + assert not d.applies + + +def test_decision_lowercases_action(): + d = HermesDelegationDecision(action="ROUTE") + assert d.action == "route" + assert d.applies + + +def test_decision_strips_invalid_reasoning_effort(): + d = HermesDelegationDecision(action="route", reasoning_effort="ultraturbo") + assert d.reasoning_effort is None + + +def test_decision_normalizes_valid_reasoning_effort_casing(): + d = HermesDelegationDecision(action="route", reasoning_effort="HIGH") + assert d.reasoning_effort == "high" + + +@pytest.mark.parametrize("effort", sorted(VALID_REASONING_EFFORTS)) +def test_decision_accepts_all_valid_reasoning_efforts(effort): + d = HermesDelegationDecision(action="route", reasoning_effort=effort) + assert d.reasoning_effort == effort + + +def test_decision_clamps_confidence_to_unit_interval(): + over = HermesDelegationDecision(confidence=5.0) + under = HermesDelegationDecision(confidence=-3.0) + assert over.confidence == 1.0 + assert under.confidence == 0.0 + + +def test_decision_handles_none_confidence(): + d = HermesDelegationDecision(confidence=None) # type: ignore[arg-type] + assert d.confidence == 0.0 + + +def test_decision_to_dict_serializes_metadata_copy(): + metadata = {"foo": "bar"} + d = HermesDelegationDecision(metadata=metadata) + out = d.to_dict() + out["metadata"]["foo"] = "MUTATED" + assert metadata["foo"] == "bar" # original untouched + + +def test_decision_to_dict_has_all_documented_keys(): + d = HermesDelegationDecision(action="route", provider="nous", model="hermes-4.1") + out = d.to_dict() + expected_keys = { + "action", + "provider", + "model", + "reasoning_effort", + "domain", + "topic", + "complexity", + "confidence", + "reason", + "source", + "metadata", + } + assert set(out) == expected_keys + + +def test_decision_applies_property(): + assert HermesDelegationDecision(action="route").applies + assert not HermesDelegationDecision(action="inherit").applies + + +def test_valid_actions_includes_both_states(): + assert VALID_ACTIONS == {"inherit", "route"}