"""The config schema: what a user's YAML must look like (2.3).
Pydantic v2 models, shape only -- no file I/O, no ``env:`` resolution, no
backend construction (those are the loader's job). Every model forbids unknown
keys (``extra="forbid"``), so a typo like ``retry_budgett`` fails loudly with
its YAML path instead of being silently ignored.
The shape mirrors the what-vs-where thesis::
run: this run's identity and inputs (name, work_dir, elements)
pipeline: how the funnel executes (stop_after, retry budget, timings)
backends: name -> {kind, options} -- the WHERE; free-form options
stages: name -> {backend, options} -- the WHAT; names its backend
A stage references a backend only by *name*; the cross-check that the name is
defined (and that ``stop_after`` names a real stage) lives on the root model,
so an inconsistent file never validates. Values may arrive as strings (the
``env:NAME`` resolution always produces strings); pydantic's lax mode coerces
numeric strings into the typed fields, while still rejecting genuinely wrong
types -- including YAML booleans where numbers belong.
"""
from __future__ import annotations
from typing import Annotated, Any
from pydantic import (
BaseModel,
ConfigDict,
Field,
StringConstraints,
field_validator,
model_validator,
)
#: A string that must not be empty (list items get no YAML path of their own,
#: so the constraint carries the message).
NonEmptyStr = Annotated[str, StringConstraints(min_length=1)]
class _Model(BaseModel):
"""Base for every config model: unknown keys are errors, never ignored."""
model_config = ConfigDict(extra="forbid")
[docs]
class RunCfg(_Model):
"""This run's identity and scientific inputs -- the ``run:`` block.
Attributes
----------
name : str
A short label for the run, shown by ``status`` and ``results``.
work_dir : str
Root of the run's on-disk workspace. The state database and every
per-material item directory live under it.
elements : list of str
The element set the compose stage enumerates over.
combination_sizes : list of int
How many elements each seed combines: ``[2]`` explores binaries,
``[2, 3]`` adds ternaries, and so on. Size ``1`` is legal and seeds
elemental solids. Omitted -- or explicitly ``[]`` -- means *every* size
from 1 to the number of distinct elements.
Enumeration grows combinatorially, so the preflight reports the seed
count before any work starts, and heavy stages can be pointed at a
remote backend when a laptop is not enough.
"""
name: NonEmptyStr
work_dir: NonEmptyStr
elements: list[NonEmptyStr] = Field(min_length=1)
# Empty = "omitted"; expanded to 1..len(elements) by _default_sizes.
# A plain comment, not a `#:` doc comment: the Attributes section above
# already documents this field, and autodoc would render both.
combination_sizes: list[int] = Field(default_factory=list)
@field_validator("combination_sizes", mode="before")
@classmethod
def _sizes_not_bool(cls, value: object) -> object:
if isinstance(value, list) and any(isinstance(size, bool) for size in value):
raise ValueError("must be integers, not booleans")
return value
@field_validator("combination_sizes")
@classmethod
def _sizes_valid(cls, value: list[int]) -> list[int]:
if any(size < 1 for size in value):
raise ValueError("each combination size must be >= 1")
return sorted(set(value))
@model_validator(mode="after")
def _default_sizes(self) -> RunCfg:
"""Expand an omitted or empty ``combination_sizes`` to all sizes 1..N.
Returns
-------
RunCfg
The model, with the field resolved. N counts *distinct* elements,
so duplicates never inflate the range, and after validation the
field is always a concrete non-empty sorted list -- nothing
downstream ever sees the empty sentinel.
"""
if not self.combination_sizes:
self.combination_sizes = list(range(1, len(set(self.elements)) + 1))
return self
[docs]
class PipelineCfg(_Model):
"""How the funnel executes -- the ``pipeline:`` block. Every key is optional.
Attributes
----------
stop_after : str or None
Name of the last stage this run executes, or ``None`` for the full
configured funnel.
A **mid-funnel stop is a pause, not a retirement**: materials rest at
that stage's output state, live and claimable, so raising or removing
``stop_after`` -- or running ``--only <stage>`` -- continues them on the
same store. Only a run that executes through the configured *last*
stage retires its materials as first-class ``DONE``.
retry_budget : int
Per-stage fault budget. A faulted material is requeued until the budget
is spent, then goes terminal ``FAILED``. Deterministic science verdicts
bypass it entirely -- no retry can change them.
lease_timeout_s : float
Age beyond which a claim is considered orphaned and released back to
its queue.
pass_interval_s : float
Sleep between orchestrator passes.
"""
stop_after: str | None = None
retry_budget: int = Field(default=3, ge=1)
lease_timeout_s: float = Field(default=900.0, gt=0)
pass_interval_s: float = Field(default=1.0, ge=0)
@field_validator("retry_budget", "lease_timeout_s", "pass_interval_s", mode="before")
@classmethod
def _reject_bool(cls, value: object) -> object:
# YAML `true` would otherwise lax-coerce to 1 / 1.0 -- the classic
# bool-is-an-int footgun. A boolean where a number belongs is a typo.
if isinstance(value, bool):
raise ValueError("must be a number, not a boolean")
return value
[docs]
class BackendCfg(_Model):
"""One compute backend -- an entry under ``backends:``, and the *where*.
Attributes
----------
kind : str
Selects the implementation through the backend registry: ``local``,
``ssh_exec``, or anything else registered.
options : dict
That kind's own settings, deliberately free-form here because each kind
needs different keys -- ``local`` wants ``work_dir`` and
``max_parallel``, ``ssh_exec`` wants ``host`` and friends. The
backend's own constructor validates what it needs and rejects what it
cannot use, which keeps this schema stable as backends grow options.
"""
kind: NonEmptyStr
options: dict[str, Any] = Field(default_factory=dict)
[docs]
class StageCfg(_Model):
"""One pipeline stage -- an entry under ``stages:``, and the *what*.
Attributes
----------
backend : str
The *name* of an entry under ``backends:``. This is the only place a
stage says where it runs, and it is one word of config rather than any
code -- moving a stage to a bigger machine is editing this line.
options : dict
Stage-specific settings -- a model path, a device, thresholds --
unpacked onto the stage's typed constructor signature, which validates
them.
"""
backend: NonEmptyStr
options: dict[str, Any] = Field(default_factory=dict)
[docs]
class Config(_Model):
"""The fully-validated configuration file -- the tool's single control surface.
Attributes
----------
run : RunCfg
Identity and scientific inputs.
pipeline : PipelineCfg
How the funnel executes.
backends : dict
The compute targets, by name.
stages : dict
The funnel stages, by name.
Notes
-----
Beyond the per-field checks, the root model enforces the two
cross-references an inconsistent file could break: every
``stages.<name>.backend`` must name a defined backend, and
``pipeline.stop_after``, when set, must name a defined stage. Every model
also forbids unknown keys, so a typo is an error rather than a setting that
silently does nothing.
"""
run: RunCfg
pipeline: PipelineCfg = Field(default_factory=PipelineCfg)
backends: dict[str, BackendCfg]
stages: dict[str, StageCfg]
@model_validator(mode="after")
def _cross_references(self) -> Config:
if not self.backends:
raise ValueError("backends: must define at least one backend")
if not self.stages:
raise ValueError("stages: must define at least one stage")
for name, stage in self.stages.items():
if stage.backend not in self.backends:
raise ValueError(
f"stages.{name}.backend: references undefined backend "
f"{stage.backend!r}; defined backends: {sorted(self.backends)}"
)
stop = self.pipeline.stop_after
if stop is not None and stop not in self.stages:
raise ValueError(
f"pipeline.stop_after: {stop!r} names no configured stage; "
f"stages: {sorted(self.stages)}"
)
return self