feat(s51): fail-fast self-contained routing config loader
Gate: pytest tests/test_hitl.py -k routing_config → 6 passed.
This commit is contained in:
parent
b50e3fdff2
commit
e9179271e6
2 changed files with 121 additions and 0 deletions
|
|
@ -35,6 +35,8 @@ from dataclasses import dataclass
|
|||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from pydantic import BaseModel, Field, model_validator
|
||||
|
||||
# Mirrored INLINE from verdicts.py (NOT imported — verdicts.py:29 pulls agent_framework). The inbox
|
||||
# predicate must match load_verdicts_from_dir EXACTLY, else pending would count as judged a file the
|
||||
# real loader drops. Kept in lockstep with verdicts._REQUIRED_VERDICT_KEYS / _INBOX_DECISION_VOCABULARY
|
||||
|
|
@ -135,6 +137,50 @@ def pending(outbox_dir: str, verdict_dir: str) -> list[PendingProposal]:
|
|||
return sorted(unjudged, key=lambda p: (p.run_id, p.verdict_id))
|
||||
|
||||
|
||||
# --- Routing config: self-contained dimension→expert table (fail-fast) ----------------------------
|
||||
# A minimal MVP stand-in for the S3.5 dimension catalog (kept DISTINCT — see the plan's Non-Goals).
|
||||
# Field names mirror ``dimension.Dimension`` so the two reconcile cleanly when S3.5 lands. No ``label``
|
||||
# field: ``_matches`` never builds a ``Dimension``, so a label would be dead single-use surface.
|
||||
|
||||
|
||||
class RoutingEntry(BaseModel):
|
||||
"""One ``dimension → expert`` routing rule. ``allowed_measure_types`` EMPTY means "any measure"
|
||||
(route by code prefix alone — see ``_matches`` + the plan's Risk #1); a non-empty set restores a
|
||||
measure gate for deployments that want one."""
|
||||
|
||||
id: str = Field(min_length=1)
|
||||
allowed_measure_types: frozenset[str] = frozenset()
|
||||
allowed_code_prefixes: frozenset[str] = frozenset()
|
||||
expert: str = Field(min_length=1)
|
||||
|
||||
|
||||
class RoutingConfig(BaseModel):
|
||||
"""The routing table. Entry ``id``s must be unique — a duplicate would make the sorted-first
|
||||
tie-break ambiguous."""
|
||||
|
||||
entries: list[RoutingEntry]
|
||||
|
||||
@model_validator(mode="after")
|
||||
def _unique_entry_ids(self) -> RoutingConfig:
|
||||
ids = [e.id for e in self.entries]
|
||||
if len(ids) != len(set(ids)):
|
||||
raise ValueError("routing config has duplicate entry ids")
|
||||
return self
|
||||
|
||||
|
||||
def load_routing_config(path: str | Path) -> RoutingConfig:
|
||||
"""Load + validate the routing config, fail-fast (mirrors ``costsim.load_pricing``): a missing
|
||||
file raises ``FileNotFoundError``; malformed data raises ``pydantic.ValidationError``; a duplicate
|
||||
entry id raises ``ValueError`` (via the after-validator). Top-level ``_``-prefixed keys are
|
||||
ignored (doc/comment convention)."""
|
||||
p = Path(path)
|
||||
if not p.is_file():
|
||||
raise FileNotFoundError(f"routing config not found: {str(p)!r}")
|
||||
raw = json.loads(p.read_text(encoding="utf-8"))
|
||||
data = {k: v for k, v in raw.items() if not k.startswith("_")}
|
||||
return RoutingConfig(**data)
|
||||
|
||||
|
||||
def _load_json_dict(file: Path) -> dict[str, Any] | None:
|
||||
"""Tolerant read: parse ``file`` as JSON and return it only if it is a dict, else ``None`` (an
|
||||
unreadable / non-JSON / non-object file is skipped by every reader here)."""
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue