from memabra.candidate_types import CandidateObject from memabra.router import FeatureScoringRouter, TaskContext def test_feature_scoring_router_computes_score_breakdown_and_selects_best(): router = FeatureScoringRouter() memory = CandidateObject( id="mem-1", type="memory", title="m1", summary="s1", confidence=0.9, success_rate=0.9, freshness=0.9, cost=0.1, risk=0.1, ) tool = CandidateObject( id="tool-1", type="tool", title="t1", summary="s1", confidence=0.8, success_rate=0.8, freshness=0.8, cost=0.1, risk=0.1, ) decision = router.choose( TaskContext(user_input="do something"), memory_candidates=[memory], skill_candidates=[], tool_candidates=[tool], ) assert decision.decision_type == "inject_memory" assert "mem-1" in decision.score_breakdown assert "tool-1" in decision.score_breakdown assert decision.score_breakdown["mem-1"] > decision.score_breakdown["tool-1"] def test_feature_scoring_router_applies_failure_penalty(): router = FeatureScoringRouter() tool_a = CandidateObject( id="tool-a", type="tool", title="ta", summary="sa", confidence=0.9, success_rate=0.9, freshness=0.9, cost=0.0, risk=0.0, ) tool_b = CandidateObject( id="tool-b", type="tool", title="tb", summary="sb", confidence=0.9, success_rate=0.9, freshness=0.9, cost=0.0, risk=0.0, ) context = TaskContext(user_input="run tool", recent_failures=["tool-b"]) decision = router.choose( context, memory_candidates=[], skill_candidates=[], tool_candidates=[tool_a, tool_b], ) assert decision.decision_type == "call_tool" assert decision.selected_ids == ["tool-a"] assert decision.score_breakdown["tool-b"] < decision.score_breakdown["tool-a"] def test_feature_scoring_router_emits_composite_action_for_preconditions(): router = FeatureScoringRouter() memory = CandidateObject( id="mem-1", type="memory", title="m1", summary="s1", confidence=0.7, success_rate=0.5, freshness=0.3, cost=0.0, risk=0.0, ) tool = CandidateObject( id="tool-1", type="tool", title="t1", summary="s1", confidence=0.9, success_rate=0.9, freshness=0.9, cost=0.0, risk=0.0, preconditions=["memory"], ) decision = router.choose( TaskContext(user_input="run tool"), memory_candidates=[memory], skill_candidates=[], tool_candidates=[tool], ) assert decision.decision_type == "composite_action" assert len(decision.composite_steps) == 2 assert decision.composite_steps[0].decision_type == "inject_memory" assert decision.composite_steps[0].selected_ids == ["mem-1"] assert decision.composite_steps[1].decision_type == "call_tool" assert decision.composite_steps[1].selected_ids == ["tool-1"] def test_feature_scoring_router_fallback_when_precondition_missing(): router = FeatureScoringRouter() tool = CandidateObject( id="tool-1", type="tool", title="t1", summary="s1", confidence=0.9, success_rate=0.9, freshness=0.9, cost=0.0, risk=0.0, preconditions=["memory"], ) decision = router.choose( TaskContext(user_input="run tool"), memory_candidates=[], skill_candidates=[], tool_candidates=[tool], ) assert decision.decision_type == "call_tool" assert decision.selected_ids == ["tool-1"]