IB-WP-0020-T02: routing config loader

build_routing_policy_from_config(config, *, workspace=None, env=None,
adapter_factory=None) materialises a parsed RoutingConfig into a live
llm-connect routing policy:

- Static RoutingPolicy when the config has no adaptive signals; one
  RoutingRule per task type, prefer = first candidate, fallback =
  second candidate (when present), max_cost_per_1k pulled from the
  preferred candidate.
- AdaptiveRoutingPolicy when default_quality_floor, any per-task
  quality_floor, or ledger_path is set. ledger_path resolves relative
  to the supplied workspace; parent directory is created so the
  ledger writes never fail on first call.
- API-key resolution from env (default os.environ) using the
  per-provider DEFAULT_API_KEY_ENV map; candidate.api_key_env overrides
  the default. Missing key raises InfospaceError("missing_routing_api_key")
  before any provider constructor runs.
- claude_code candidates need no API key (shells out to the local CLI).
- adapter_factory hook lets tests inject a sentinel-returning factory
  so policy construction stays network- and llm-adapter-free.

Eight new tests cover: static-policy default, adaptive selection via
ledger_path, adaptive selection via quality_floor, multi-candidate
fallback rule, real-factory smoke (OpenRouterAdapter constructed with
env API key), missing-key fast-fail, claude_code zero-key path, and
custom api_key_env override.

168 tests pass, 1 skipped.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
2026-05-18 19:58:15 +02:00
parent c11a942bb7
commit 82468c2165
3 changed files with 324 additions and 2 deletions

View File

@@ -261,6 +261,189 @@ def test_rejects_non_mapping_stage_map() -> None:
assert exc_info.value.code == "invalid_routing_stage_map"
# ---------------------------------------------------------------------------
# T02 — loader that materialises a config into a live llm-connect policy
# ---------------------------------------------------------------------------
def _fake_adapter_factory_record(record: list):
"""Return a factory that records calls and returns a sentinel string."""
def _factory(candidate, env):
record.append({"id": candidate.id, "provider": candidate.provider, "model": candidate.model})
return f"adapter:{candidate.id}"
return _factory
def test_build_routing_policy_returns_static_when_no_adaptive_signals() -> None:
from llm_connect.routing import RoutingPolicy
from infospace_bench.routing_config import build_routing_policy_from_config
config = parse_routing_config(MINIMAL)
calls: list[dict] = []
policy = build_routing_policy_from_config(
config, adapter_factory=_fake_adapter_factory_record(calls)
)
assert isinstance(policy, RoutingPolicy)
assert type(policy).__name__ == "RoutingPolicy", "no adaptive signals -> static policy"
assert len(policy.rules) == 1
assert policy.rules[0].task_type == "summarize-source"
assert policy.rules[0].prefer == "adapter:openrouter:gpt-4o-mini"
assert policy.rules[0].fallback is None
assert calls and calls[0]["provider"] == "openrouter"
def test_build_routing_policy_returns_adaptive_when_ledger_path_set(tmp_path: Path) -> None:
from llm_connect.routing import AdaptiveRoutingPolicy
from infospace_bench.routing_config import build_routing_policy_from_config
data = {
**MINIMAL,
"ledger_path": "output/routing/quality.jsonl",
}
config = parse_routing_config(data)
policy = build_routing_policy_from_config(
config,
workspace=tmp_path,
adapter_factory=_fake_adapter_factory_record([]),
)
assert isinstance(policy, AdaptiveRoutingPolicy)
assert policy.ledger is not None
expected_path = tmp_path / "output" / "routing" / "quality.jsonl"
assert Path(policy.ledger.path) == expected_path
def test_build_routing_policy_returns_adaptive_when_quality_floor_set() -> None:
from llm_connect.routing import AdaptiveRoutingPolicy
from infospace_bench.routing_config import build_routing_policy_from_config
data = {
**MINIMAL,
"default_quality_floor": 0.8,
}
config = parse_routing_config(data)
policy = build_routing_policy_from_config(
config, adapter_factory=_fake_adapter_factory_record([])
)
assert isinstance(policy, AdaptiveRoutingPolicy)
assert policy.ledger is None # no ledger_path set
def test_build_routing_policy_routes_fallback_for_multi_candidate_rule() -> None:
from infospace_bench.routing_config import build_routing_policy_from_config
data = {
"schema_version": 1,
"task_types": {
"extract-entities": {
"candidates": [
{
"id": "openrouter:cheap",
"provider": "openrouter",
"model": "openai/gpt-4o-mini",
"max_cost_per_1k": 0.001,
},
{
"id": "openrouter:smart",
"provider": "openrouter",
"model": "anthropic/claude-3.5-sonnet",
},
],
},
},
}
config = parse_routing_config(data)
policy = build_routing_policy_from_config(
config, adapter_factory=_fake_adapter_factory_record([])
)
rule = policy.rules[0]
assert rule.prefer == "adapter:openrouter:cheap"
assert rule.max_cost_per_1k == 0.001
assert rule.fallback == "adapter:openrouter:smart"
def test_build_routing_policy_resolves_api_key_from_env() -> None:
from infospace_bench.routing_config import (
build_routing_policy_from_config,
_default_adapter_factory,
)
config = parse_routing_config(MINIMAL)
# Smoke: real factory with a fake env should construct an OpenRouterAdapter.
env = {"OPENROUTER_API_KEY": "sk-fake-test-key"}
policy = build_routing_policy_from_config(config, env=env)
rule = policy.rules[0]
# The constructed adapter is an OpenRouterAdapter from llm-connect.
from llm_connect.openrouter import OpenRouterAdapter
assert isinstance(rule.prefer, OpenRouterAdapter)
def test_build_routing_policy_fails_fast_on_missing_api_key() -> None:
from infospace_bench.routing_config import build_routing_policy_from_config
config = parse_routing_config(MINIMAL)
# Empty env — the candidate's required env var is unset.
with pytest.raises(InfospaceError) as exc_info:
build_routing_policy_from_config(config, env={})
assert exc_info.value.code == "missing_routing_api_key"
assert exc_info.value.detail["provider"] == "openrouter"
def test_build_routing_policy_claude_code_needs_no_api_key() -> None:
from infospace_bench.routing_config import build_routing_policy_from_config
from llm_connect.claude_code import ClaudeCodeAdapter
data = {
"schema_version": 1,
"task_types": {
"baseline": {
"candidates": [
{"id": "claude-code", "provider": "claude_code", "model": "claude-opus-4-7"},
],
},
},
}
config = parse_routing_config(data)
policy = build_routing_policy_from_config(config, env={})
assert isinstance(policy.rules[0].prefer, ClaudeCodeAdapter)
def test_build_routing_policy_honours_custom_api_key_env() -> None:
from infospace_bench.routing_config import build_routing_policy_from_config
from llm_connect.openrouter import OpenRouterAdapter
data = {
"schema_version": 1,
"task_types": {
"summarize-source": {
"candidates": [
{
"id": "openrouter:gpt-4o-mini",
"provider": "openrouter",
"model": "openai/gpt-4o-mini",
"api_key_env": "ALT_OPENROUTER_KEY",
},
],
},
},
}
config = parse_routing_config(data)
with pytest.raises(InfospaceError) as exc_info:
build_routing_policy_from_config(config, env={"OPENROUTER_API_KEY": "wrong-default"})
assert exc_info.value.code == "missing_routing_api_key"
assert exc_info.value.detail["api_key_env"] == "ALT_OPENROUTER_KEY"
policy = build_routing_policy_from_config(
config, env={"ALT_OPENROUTER_KEY": "sk-fake"}
)
assert isinstance(policy.rules[0].prefer, OpenRouterAdapter)
def test_rejects_non_string_ledger_path() -> None:
payload = {
"schema_version": 1,