135 lines
5.1 KiB
Python
135 lines
5.1 KiB
Python
"""Fail-fast contract loaders (brief NFR: validate ALL configs at startup, before any
|
|
chat-client is constructed or called).
|
|
|
|
Four Pydantic contracts give JSON-Schema-grade validation (CLAUDE.md convention):
|
|
|
|
* ``DataSourceContract`` — the local-folder data source (docs dir + top_k).
|
|
* ``ModelMapContract`` — validates ``data/model_map.json`` (Step 8): a role->model map per
|
|
backend ``Profile``, each with a ``default``.
|
|
* ``TerminationContract`` — the stop criteria + budget cap (max_rounds, max_tokens) required
|
|
at startup (never an unbounded loop).
|
|
* ``FeedbackContract`` — the expert-verdict feedback shape (decision + rationale).
|
|
|
|
``load_contracts`` validates all four and raises ``pydantic.ValidationError`` on the first
|
|
malformed one — purely at the config layer, so it can run before any backend/client exists.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from importlib.resources import files
|
|
from pathlib import Path
|
|
from typing import Any, Literal
|
|
|
|
from pydantic import BaseModel, Field, model_validator
|
|
|
|
from portfolio_optimiser.backends import Profile
|
|
|
|
_MODEL_MAP_RESOURCE = "data/model_map.json"
|
|
|
|
|
|
class DataSourceContract(BaseModel):
|
|
"""The local-folder data source config (JSON-Schema-validated, fail-fast)."""
|
|
|
|
docs_dir: str = Field(min_length=1)
|
|
top_k: int = Field(gt=0)
|
|
|
|
|
|
class ModelMapContract(BaseModel):
|
|
"""Role -> model/deployment map per backend profile (validates data/model_map.json)."""
|
|
|
|
local: dict[str, str] = Field(min_length=1)
|
|
azure: dict[str, str] = Field(min_length=1)
|
|
|
|
@model_validator(mode="after")
|
|
def _each_profile_has_default(self) -> ModelMapContract:
|
|
for prof in (Profile.LOCAL.value, Profile.AZURE.value):
|
|
if "default" not in getattr(self, prof):
|
|
raise ValueError(f"model_map.{prof} must include a 'default' model id")
|
|
return self
|
|
|
|
|
|
class TerminationContract(BaseModel):
|
|
"""Stop criteria + budget cap required at startup (fail-fast, never unbounded)."""
|
|
|
|
max_rounds: int = Field(gt=0)
|
|
max_tokens: int = Field(gt=0)
|
|
|
|
|
|
class FeedbackContract(BaseModel):
|
|
"""The expert-verdict feedback shape fed back into the VerdictStore (Layer-2)."""
|
|
|
|
decision: Literal["approved", "rejected"]
|
|
rationale: str = Field(min_length=1)
|
|
|
|
|
|
class GoalContract(BaseModel):
|
|
"""A configurable savings goal for a portfolio run: an absolute *øre* target and/or a percent
|
|
target, in ``hard`` or ``soft`` mode (default hard — brief §4.3). At least one of ``absolute_ore``
|
|
/ ``percent`` must be set. Distinct from ``TerminationContract`` (the token/round budget cap):
|
|
this is a domain GOAL (savings reached), not resource exhaustion."""
|
|
|
|
absolute_ore: int | None = Field(default=None, ge=0)
|
|
percent: float | None = Field(default=None, ge=0, le=100)
|
|
mode: Literal["hard", "soft"] = "hard"
|
|
|
|
@model_validator(mode="after")
|
|
def _at_least_one_target(self) -> GoalContract:
|
|
if self.absolute_ore is None and self.percent is None:
|
|
raise ValueError("GoalContract requires at least one of absolute_ore or percent")
|
|
return self
|
|
|
|
|
|
class GoalConfig(BaseModel):
|
|
"""Both goal levels at once: an optional portfolio-wide goal plus per-project goals keyed by
|
|
``project_id`` — so 'this project's goal stops THAT project' is unambiguously keyed (SC6 exercises
|
|
both branches together)."""
|
|
|
|
portfolio: GoalContract | None = None
|
|
per_project: dict[str, GoalContract] = Field(default_factory=dict)
|
|
|
|
|
|
def load_goal_config(path: str) -> GoalConfig:
|
|
"""Fail-fast standalone loader (mirrors ``okf.load_ir_projection``): a missing file raises
|
|
``FileNotFoundError``, a malformed config raises ``pydantic.ValidationError``. Deliberately NOT
|
|
folded into ``load_contracts`` — that would break its positional callers (``run.py:223``); the
|
|
goal config is loaded separately at the orchestration entry point (Step 8)."""
|
|
p = Path(path)
|
|
if not p.is_file():
|
|
raise FileNotFoundError(f"goal config not found: {path!r}")
|
|
data = json.loads(p.read_text(encoding="utf-8"))
|
|
return GoalConfig(**data)
|
|
|
|
|
|
class Contracts(BaseModel):
|
|
"""The validated bundle of all startup contracts."""
|
|
|
|
data_source: DataSourceContract
|
|
model_map: ModelMapContract
|
|
termination: TerminationContract
|
|
feedback: FeedbackContract
|
|
|
|
|
|
def _bundled_model_map() -> dict[str, Any]:
|
|
return json.loads(
|
|
files("portfolio_optimiser").joinpath(_MODEL_MAP_RESOURCE).read_text(encoding="utf-8")
|
|
)
|
|
|
|
|
|
def load_contracts(
|
|
data_source: dict[str, Any],
|
|
termination: dict[str, Any],
|
|
feedback: dict[str, Any],
|
|
*,
|
|
model_map: dict[str, Any] | None = None,
|
|
) -> Contracts:
|
|
"""Validate ALL contracts at startup (fail-fast, before any chat-client is built). Raises
|
|
``pydantic.ValidationError`` on the first malformed contract. ``model_map`` defaults to the
|
|
bundled ``data/model_map.json`` (the same file Step 8 ships)."""
|
|
raw_map = _bundled_model_map() if model_map is None else model_map
|
|
return Contracts(
|
|
data_source=DataSourceContract(**data_source),
|
|
model_map=ModelMapContract(**raw_map),
|
|
termination=TerminationContract(**termination),
|
|
feedback=FeedbackContract(**feedback),
|
|
)
|