reach.config

Configuration models, settings resolution, and path expansions.

Define configuration models, path expansion, and deterministic digest generation.

AgentProfile

Bases: BaseModel

Configuration profile for a named agent runtime in reach.toml.

Source code in src/reach/config.py

class AgentProfile(BaseModel):
    """Configuration profile for a named agent runtime in reach.toml."""

    model_config = ConfigDict(frozen=True, extra="forbid")

    default_model: str | None = None
    default_provider: str | None = None
    skills_dir: str | None = None
    user_skills_dir: str | None = None
    models: tuple[str, ...] = ()

CatalogSettings

Bases: BaseModel

Configuration settings for catalog composition.

Source code in src/reach/config.py

class CatalogSettings(BaseModel):
    """Configuration settings for catalog composition."""

    model_config = ConfigDict(frozen=True, extra="forbid")

    mode: CatalogMode = CatalogMode.NEIGHBORHOOD
    size: int = Field(default=20, ge=2)
    rivals: int = Field(default=10, ge=1)
    seed: int = 0
    scorer: str = "hybrid"

CheckSettings

Bases: BaseModel

Configuration settings for CI/CD regression checks and quality gates.

Source code in src/reach/config.py

class CheckSettings(BaseModel):
    """Configuration settings for CI/CD regression checks and quality gates."""

    model_config = ConfigDict(frozen=True, extra="forbid")

    min_recall: float = Field(default=0.80, ge=0.0, le=1.0)
    min_accuracy: float = Field(default=0.80, ge=0.0, le=1.0)
    max_misroute: float = Field(default=0.10, ge=0.0, le=1.0)
    min_entrypoint: float | None = Field(default=None, ge=0.0, le=1.0)
    min_reachability: float | None = Field(default=None, ge=0.0, le=1.0)
    min_efficiency: float | None = Field(default=None, ge=0.0, le=1.0)
    min_f1: float | None = Field(default=None, ge=0.0, le=1.0)
    max_redundancy: float | None = Field(default=None, ge=0.0)
    budget: int = Field(default=50, ge=1)
    strict: bool = True
    since: str = "HEAD~1"

DiffSettings

Bases: BaseModel

Configuration settings for A/B evaluation diffing and noise floor calibration.

Source code in src/reach/config.py

class DiffSettings(BaseModel):
    """Configuration settings for A/B evaluation diffing and noise floor calibration."""

    model_config = ConfigDict(frozen=True, extra="forbid")

    confidence: float = Field(default=DEFAULT_CONFIDENCE, gt=0.0, lt=1.0)
    power: float = Field(default=DEFAULT_POWER, gt=0.0, lt=1.0)
    noise_inflation: float = Field(default=1.265, gt=0.0)

    @property
    def over_dispersion(self) -> float:
        """Derive the variance over-dispersion ratio from noise_inflation squared."""
        return self.noise_inflation**2

over_dispersion property

over_dispersion: float

Derive the variance over-dispersion ratio from noise_inflation squared.

DiscoverySettings

Bases: BaseModel

Configuration settings for skill corpus discovery precedence.

Source code in src/reach/config.py

class DiscoverySettings(BaseModel):
    """Configuration settings for skill corpus discovery precedence."""

    model_config = ConfigDict(frozen=True, extra="forbid")

    precedence: tuple[str, ...] = DEFAULT_DISCOVERY_PRECEDENCE

GeneralSettings

Bases: BaseModel

Global default settings from reach.toml.

Source code in src/reach/config.py

class GeneralSettings(BaseModel):
    """Global default settings from reach.toml."""

    model_config = ConfigDict(frozen=True, extra="forbid")

    default_agent: str = "antigravity-cli"

LintSettings

Bases: BaseModel

Configuration settings for static skill linting and validation thresholds.

Source code in src/reach/config.py

class LintSettings(BaseModel):
    """Configuration settings for static skill linting and validation thresholds."""

    model_config = ConfigDict(frozen=True, extra="forbid")

    max_description_length: Annotated[int, Field(ge=1)] = 1024
    max_name_length: Annotated[int, Field(ge=1)] = 64
    min_description_length: Annotated[int, Field(ge=1)] = 20
    catalog_budget_chars: Annotated[int, Field(ge=1)] | None = DEFAULT_CATALOG_BUDGET_CHARS
    similarity_threshold: Annotated[float, Field(ge=0.0, le=1.0)] = 0.92
    mutual_handoff_similarity_threshold: Annotated[float, Field(ge=0.0, le=1.0)] = 0.75
    mutual_handoff_lexical_threshold: Annotated[float, Field(ge=0.0, le=1.0)] = 0.35
    rules: dict[str, RuleSeverity] = Field(default_factory=dict)

    def with_rule_overrides(
        self,
        overrides: Mapping[str, Any] | None = None,
    ) -> LintSettings:
        """Return a validated copy of LintSettings with rule severity overrides applied."""
        if not overrides:
            return self
        merged_rules = {**self.rules, **overrides}
        validated_rules = (
            type(self).model_validate({**self.model_dump(), "rules": merged_rules}).rules
        )
        return self.model_copy(update={"rules": validated_rules})

    @classmethod
    def from_settings(
        cls,
        settings: Mapping[str, object] | None = None,
        overrides: Mapping[str, Any] | None = None,
    ) -> LintSettings:
        """Construct a LintSettings from loaded reach.toml settings and CLI overrides."""
        if settings is None:
            settings = load_config()

        lint_section = settings.get("lint", {}) if isinstance(settings, Mapping) else {}
        retrieval_section = settings.get("retrieval", {}) if isinstance(settings, Mapping) else {}
        catalog_section = settings.get("catalog", {}) if isinstance(settings, Mapping) else {}

        payload: dict[str, Any] = dict(lint_section) if isinstance(lint_section, Mapping) else {}
        raw_rules = payload.get("rules")
        rules: dict[str, Any] = dict(raw_rules) if isinstance(raw_rules, Mapping) else {}
        if overrides:
            rules.update(overrides)
        if rules or "rules" in payload:
            payload["rules"] = rules

        lint_base = cls.model_validate(payload)
        catalog_cfg = CatalogSettings.model_validate(
            catalog_section if isinstance(catalog_section, Mapping) else {}
        )
        retrieval_cfg = RetrievalSettings.model_validate(
            retrieval_section if isinstance(retrieval_section, Mapping) else {}
        )
        return _apply_lint_inheritance(lint_base, catalog_cfg, retrieval_cfg)

from_settings classmethod

from_settings(
    settings: Mapping[str, object] | None = None,
    overrides: Mapping[str, Any] | None = None,
) -> LintSettings

Construct a LintSettings from loaded reach.toml settings and CLI overrides.

Source code in src/reach/config.py

@classmethod
def from_settings(
    cls,
    settings: Mapping[str, object] | None = None,
    overrides: Mapping[str, Any] | None = None,
) -> LintSettings:
    """Construct a LintSettings from loaded reach.toml settings and CLI overrides."""
    if settings is None:
        settings = load_config()

    lint_section = settings.get("lint", {}) if isinstance(settings, Mapping) else {}
    retrieval_section = settings.get("retrieval", {}) if isinstance(settings, Mapping) else {}
    catalog_section = settings.get("catalog", {}) if isinstance(settings, Mapping) else {}

    payload: dict[str, Any] = dict(lint_section) if isinstance(lint_section, Mapping) else {}
    raw_rules = payload.get("rules")
    rules: dict[str, Any] = dict(raw_rules) if isinstance(raw_rules, Mapping) else {}
    if overrides:
        rules.update(overrides)
    if rules or "rules" in payload:
        payload["rules"] = rules

    lint_base = cls.model_validate(payload)
    catalog_cfg = CatalogSettings.model_validate(
        catalog_section if isinstance(catalog_section, Mapping) else {}
    )
    retrieval_cfg = RetrievalSettings.model_validate(
        retrieval_section if isinstance(retrieval_section, Mapping) else {}
    )
    return _apply_lint_inheritance(lint_base, catalog_cfg, retrieval_cfg)

with_rule_overrides

with_rule_overrides(
    overrides: Mapping[str, Any] | None = None,
) -> LintSettings

Return a validated copy of LintSettings with rule severity overrides applied.

Source code in src/reach/config.py

def with_rule_overrides(
    self,
    overrides: Mapping[str, Any] | None = None,
) -> LintSettings:
    """Return a validated copy of LintSettings with rule severity overrides applied."""
    if not overrides:
        return self
    merged_rules = {**self.rules, **overrides}
    validated_rules = (
        type(self).model_validate({**self.model_dump(), "rules": merged_rules}).rules
    )
    return self.model_copy(update={"rules": validated_rules})

OptimizeSettings

Bases: BaseModel

Configuration settings for automated skill description optimization.

Source code in src/reach/config.py

class OptimizeSettings(BaseModel):
    """Configuration settings for automated skill description optimization."""

    model_config = ConfigDict(frozen=True, extra="forbid")

    budget: int = Field(default=30, ge=1)
    temperature: float = Field(default=0.7, ge=0.0, le=2.0)
    iterations: int = Field(default=1, ge=1, le=10)
    holdout: float = Field(default=0.2, ge=0.0, le=0.9)
    review: bool = Field(default=False)
    auto_queries: bool = Field(default=True)
    adversarial_count: int = Field(default=5, ge=0)
    positive_count: int = Field(default=5, ge=1)
    seed: int = Field(default=42)
    review_timeout: float = Field(default=600.0, gt=0.0)
    workers: int = Field(default=4, ge=1)
    with_handoff: bool = Field(default=False)

OverlapSettings

Bases: BaseModel

Configuration settings for lexical overlap and vocabulary rewrite heuristics.

Source code in src/reach/config.py

class OverlapSettings(BaseModel):
    """Configuration settings for lexical overlap and vocabulary rewrite heuristics."""

    model_config = ConfigDict(frozen=True, extra="forbid")

    contender_band: Annotated[float, Field(ge=0.0, le=1.0)] = 0.90
    material_share: Annotated[float, Field(ge=0.0, le=1.0)] = 0.01
    claim_limit: Annotated[int, Field(ge=1)] = 8
    min_claim_length: Annotated[int, Field(ge=1)] = 3
    min_claim_uses: Annotated[int, Field(ge=1)] = 2

PlanSettings

Bases: BaseModel

Configuration settings for probe planning and retry policies.

Source code in src/reach/config.py

class PlanSettings(BaseModel):
    """Configuration settings for probe planning and retry policies."""

    model_config = ConfigDict(frozen=True, extra="forbid")

    attempts: int = Field(default=DEFAULT_ATTEMPTS, ge=1)
    retries: int = Field(default=2, ge=0)
    backoff_s: float = Field(default=5.0, ge=0)
    pause_s: float = Field(default=0.0, ge=0)
    workers: int = Field(default=1, ge=1)

    def resolve_sweep_attempts(self, cli_attempts: int | None = None) -> int:
        """Return effective sweep attempts, defaulting to 1 unless explicitly configured."""
        if cli_attempts is not None:
            if cli_attempts < 1:
                msg = f"Sweep attempts must be at least 1, got {cli_attempts}"
                raise ValueError(msg)
            return cli_attempts
        return self.attempts if "attempts" in self.model_fields_set else 1

    def resolve_eval_attempts(
        self,
        agent: str,
        cli_attempts: int | None = None,
        *,
        quick: bool = False,
    ) -> int:
        """Return effective eval attempts, defaulting to 1 for deterministic keyword runs."""
        if cli_attempts is not None:
            if cli_attempts < 1:
                msg = f"Eval attempts must be at least 1, got {cli_attempts}"
                raise ValueError(msg)
            return cli_attempts
        if "attempts" in self.model_fields_set:
            return self.attempts
        if agent == KEYWORD_AGENT:
            return 1
        return QUICK_ATTEMPTS if quick else self.attempts

resolve_eval_attempts

resolve_eval_attempts(
    agent: str,
    cli_attempts: int | None = None,
    *,
    quick: bool = False,
) -> int

Return effective eval attempts, defaulting to 1 for deterministic keyword runs.

Source code in src/reach/config.py

def resolve_eval_attempts(
    self,
    agent: str,
    cli_attempts: int | None = None,
    *,
    quick: bool = False,
) -> int:
    """Return effective eval attempts, defaulting to 1 for deterministic keyword runs."""
    if cli_attempts is not None:
        if cli_attempts < 1:
            msg = f"Eval attempts must be at least 1, got {cli_attempts}"
            raise ValueError(msg)
        return cli_attempts
    if "attempts" in self.model_fields_set:
        return self.attempts
    if agent == KEYWORD_AGENT:
        return 1
    return QUICK_ATTEMPTS if quick else self.attempts

resolve_sweep_attempts

resolve_sweep_attempts(
    cli_attempts: int | None = None,
) -> int

Return effective sweep attempts, defaulting to 1 unless explicitly configured.

Source code in src/reach/config.py

def resolve_sweep_attempts(self, cli_attempts: int | None = None) -> int:
    """Return effective sweep attempts, defaulting to 1 unless explicitly configured."""
    if cli_attempts is not None:
        if cli_attempts < 1:
            msg = f"Sweep attempts must be at least 1, got {cli_attempts}"
            raise ValueError(msg)
        return cli_attempts
    return self.attempts if "attempts" in self.model_fields_set else 1

QuerySettings

Bases: BaseModel

Configuration settings for query synthesis, difficulty, and leakage detection.

Source code in src/reach/config.py

class QuerySettings(BaseModel):
    """Configuration settings for query synthesis, difficulty, and leakage detection."""

    model_config = ConfigDict(frozen=True, extra="forbid")

    count: int = Field(default=3, ge=1)
    adversarial_count: int = Field(default=1, ge=0)
    top_rivals: int = Field(default=3, ge=0)
    distinctive_idf_floor: float = Field(default=0.693147, ge=0.0)

RegistrySettings

Bases: BaseModel

Configuration settings for Google Cloud Agent Registry.

Source code in src/reach/config.py

class RegistrySettings(BaseModel):
    """Configuration settings for Google Cloud Agent Registry."""

    model_config = ConfigDict(frozen=True, extra="forbid")

    project: str | None = None
    location: str = "global"
    publisher: str | None = None
    cache_ttl_seconds: int = 300
    registry: bool = False
    fresh: bool = False
    no_cache: bool = False

RetrievalSettings

Bases: BaseModel

Configuration settings for dense, lexical, and hybrid retrieval.

Source code in src/reach/config.py

class RetrievalSettings(BaseModel):
    """Configuration settings for dense, lexical, and hybrid retrieval."""

    model_config = ConfigDict(frozen=True, extra="forbid")

    scorer: str = Field(default="hybrid")
    model: str = Field(default="minishlab/potion-retrieval-32M")
    rrf_k: int = Field(default=60, gt=0)
    similarity_threshold: float = Field(default=0.92, ge=0.0, le=1.0)
    bm25_k1: float = Field(default=1.5, gt=0.0)
    bm25_b: float = Field(default=0.75, ge=0.0, le=1.0)

RunConfig

Bases: BaseModel

Encapsulate all parameters required to drive an evaluation run.

Source code in src/reach/config.py

class RunConfig(BaseModel):
    """Encapsulate all parameters required to drive an evaluation run."""

    model_config = ConfigDict(frozen=True, extra="forbid")

    general: GeneralSettings = Field(default_factory=GeneralSettings)
    catalog: CatalogSettings = Field(default_factory=CatalogSettings)
    runtime: RuntimeSettings = Field(default_factory=RuntimeSettings)
    plan: PlanSettings = Field(default_factory=PlanSettings)
    study: StudySettings = Field(default_factory=StudySettings)
    lint: LintSettings = Field(default_factory=LintSettings)
    check: CheckSettings = Field(default_factory=CheckSettings)
    retrieval: RetrievalSettings = Field(default_factory=RetrievalSettings)
    discovery: DiscoverySettings = Field(default_factory=DiscoverySettings)
    overlap: OverlapSettings = Field(default_factory=OverlapSettings)
    diff: DiffSettings = Field(default_factory=DiffSettings)
    query: QuerySettings = Field(default_factory=QuerySettings)
    optimize: OptimizeSettings = Field(default_factory=OptimizeSettings)
    registry: RegistrySettings = Field(default_factory=RegistrySettings)

    @model_validator(mode="before")
    @classmethod
    def _strip_registry_tables(cls, data: object) -> object:
        """Validate and strip top-level agent and model registry tables merged by load_config()."""
        if isinstance(data, Mapping) and ("agents" in data or "models" in data):
            if "agents" in data:
                _AGENT_REGISTRY_ADAPTER.validate_python(data["agents"])
            if "models" in data:
                _model_registry_adapter().validate_python(data["models"])
            return {k: v for k, v in data.items() if k not in {"agents", "models"}}
        return data

    @model_validator(mode="after")
    def _inherit_optimize_workers_from_plan(self) -> Self:
        """Inherit plan.workers into optimize.workers when optimize.workers is unset."""
        if (
            "workers" not in self.optimize.model_fields_set
            and "workers" in self.plan.model_fields_set
        ):
            object.__setattr__(
                self,
                "optimize",
                self.optimize.model_copy(update={"workers": self.plan.workers}),
            )
        return self

    @model_validator(mode="after")
    def _inherit_runtime_agent_from_general(self) -> Self:
        """Inherit general.default_agent into runtime.agent when runtime.agent is unset."""
        if "agent" not in self.runtime.model_fields_set and self.general.default_agent:
            object.__setattr__(
                self,
                "runtime",
                self.runtime.model_copy(update={"agent": self.general.default_agent}),
            )
        return self

    @model_validator(mode="after")
    def _inherit_lint_settings(self) -> Self:
        """Inherit retrieval threshold and scoped catalog mode into lint settings."""
        inherited = _apply_lint_inheritance(self.lint, self.catalog, self.retrieval)
        if inherited is not self.lint:
            object.__setattr__(self, "lint", inherited)
        return self

    def require_queries(self, hint: str = "") -> Path:
        """Forward queries path requirement to study settings."""
        return self.study.require_queries(hint)

    def require_skills(self, hint: str = "") -> Path:
        """Forward skills path requirement to study settings."""
        return self.study.require_skills(hint)

    def require_workdir(self, hint: str = "") -> Path:
        """Forward workdir path requirement to study settings."""
        return self.study.require_workdir(hint)

    _SECTION_MAP: ClassVar[dict[type[BaseModel], str]] = {
        GeneralSettings: "general",
        CatalogSettings: "catalog",
        RuntimeSettings: "runtime",
        PlanSettings: "plan",
        StudySettings: "study",
        LintSettings: "lint",
        CheckSettings: "check",
        RetrievalSettings: "retrieval",
        DiscoverySettings: "discovery",
        OverlapSettings: "overlap",
        DiffSettings: "diff",
        QuerySettings: "query",
        OptimizeSettings: "optimize",
        RegistrySettings: "registry",
    }

    def get_section[T: BaseModel](self, section_cls: type[T]) -> T:
        """Retrieve the configuration section corresponding to section_cls."""
        attr_name = self._SECTION_MAP.get(section_cls)
        if attr_name is not None and hasattr(self, attr_name):
            val = getattr(self, attr_name)
            if isinstance(val, section_cls):
                return val
        for attr in self.__dict__:
            val = getattr(self, attr)
            if isinstance(val, section_cls):
                return val
        return section_cls()

    def resolve_section[T: BaseModel](
        self,
        section_cls: type[T],
        *,
        explicit_settings: T | None = None,
        **overrides: object,
    ) -> T:
        """Resolve effective settings section against this RunConfig instance."""
        return self.resolve(
            section_cls,
            config=self,
            explicit_settings=explicit_settings,
            **overrides,
        )

    @classmethod
    def resolve[T: BaseModel](
        cls,
        section_cls: type[T],
        config: RunConfig | None = None,
        *,
        explicit_settings: T | None = None,
        **overrides: object,
    ) -> T:
        """Resolve effective settings section from an optional RunConfig or defaults."""
        section = config.get_section(section_cls) if config is not None else None
        active_overrides = dict(overrides)

        if issubclass(section_cls, RuntimeSettings):
            configured_runtime = section if isinstance(section, RuntimeSettings) else None
            base_runtime = (
                explicit_settings
                if isinstance(explicit_settings, RuntimeSettings)
                else (configured_runtime if configured_runtime is not None else RuntimeSettings())
            )
            resolved_runtime = base_runtime.with_overrides(**active_overrides)
            if config is not None:
                resolved_runtime = _propagate_registry_to_runtime(resolved_runtime, config.registry)
            return cast(T, resolved_runtime)

        resolved = resolve_sub_settings(
            section_cls,
            config_section=section,
            explicit_settings=explicit_settings,
            **active_overrides,
        )

        if isinstance(resolved, RegistrySettings):
            return cast(T, cls._resolve_registry_section(resolved, config))
        return cast(T, resolved)

    @staticmethod
    def _resolve_registry_section(
        resolved: RegistrySettings,
        config: RunConfig | None,
    ) -> RegistrySettings:
        """Resolve environment and TOML fallbacks for a RegistrySettings section."""
        if resolved.project:
            project: str | None = resolved.project.strip()
        elif config is not None:
            project = _env_google_project()
        else:
            project = resolve_registry_project(None)

        explicit_loc = resolved.location if "location" in resolved.model_fields_set else None
        if explicit_loc:
            location = explicit_loc.strip()
        elif config is not None:
            location = _env_google_location()
        else:
            location = resolve_registry_location(None)
        return resolved.model_copy(update={"project": project, "location": location})

    @classmethod
    def from_toml(
        cls,
        path: Path | str | None = None,
        *,
        skills: Path | str | None = None,
    ) -> RunConfig:
        """Load and validate a RunConfig from a TOML configuration file."""
        resolved = (
            _discover_config_path(None)[0] if path is None else Path(path).expanduser().resolve()
        )
        if resolved is None:
            config = cls()
            if skills is not None:
                resolved_skills = expand_path(skills, base=Path.cwd())
                config = config.model_copy(
                    update={"study": config.study.model_copy(update={"skills": resolved_skills})}
                )
            return config

        payload = tomllib.loads(resolved.read_text(encoding="utf-8"))
        study = dict(payload.get("study", {}))
        if skills is not None:
            study["skills"] = skills
        for key in ("skills", "queries", "workdir", "out"):
            raw = study.get(key)
            if raw is not None:
                study[key] = expand_path(raw, base=resolved.parent)
        payload["study"] = study
        return cls.model_validate(payload)

    @staticmethod
    def declared(path: Path | str, section: str, key: str) -> bool:
        """Check whether a setting key is explicitly declared in a TOML file."""
        payload = tomllib.loads(
            Path(path).expanduser().resolve().read_text(encoding="utf-8"),
        )
        return key in payload.get(section, {})

    def with_overrides(self, **sections: dict[str, Any]) -> RunConfig:
        """Return copy of configuration with section field overrides applied."""
        payload: dict[str, Any] = {}
        for section_name in type(self).model_fields:
            section_model = getattr(self, section_name)
            if isinstance(section_model, BaseModel):
                sec_dump = {
                    **section_model.model_dump(exclude_defaults=True),
                    **section_model.model_dump(exclude_unset=True),
                }
                if section_name == "runtime" and "agent" not in section_model.model_fields_set:
                    sec_dump.pop("agent", None)
                if section_name == "optimize" and "workers" not in section_model.model_fields_set:
                    sec_dump.pop("workers", None)
                if isinstance(section_model, LintSettings):
                    if (
                        self.catalog.mode in {CatalogMode.NEIGHBORHOOD, CatalogMode.SINGLETON}
                        and section_model.catalog_budget_chars is None
                    ):
                        sec_dump.pop("catalog_budget_chars", None)
                    if (
                        "similarity_threshold" in self.retrieval.model_fields_set
                        and section_model.similarity_threshold
                        == self.retrieval.similarity_threshold
                    ):
                        sec_dump.pop("similarity_threshold", None)
                payload[section_name] = sec_dump
            else:
                payload[section_name] = section_model

        for section, fields in sections.items():
            live = {k: v for k, v in (fields or {}).items() if v is not None}
            if section == "runtime":
                new_runtime = self.runtime.with_overrides(**live)
                sec_dump = {
                    **new_runtime.model_dump(exclude_defaults=True),
                    **new_runtime.model_dump(exclude_unset=True),
                }
                if "agent" not in new_runtime.model_fields_set:
                    sec_dump.pop("agent", None)
                payload["runtime"] = sec_dump
            else:
                payload[section] = {**payload.get(section, {}), **live}
        return type(self).model_validate(payload)

    @cached_property
    def _digests(self) -> Digests:
        """Compute and memoize fingerprint, arm, and condition digests."""
        return digest_material(self.material())

    @property
    def fingerprint(self) -> str:
        """Return deterministic configuration fingerprint."""
        return self._digests.fingerprint

    def material(self) -> dict[str, Any]:
        """Dump configuration dictionary with agent options resolved to defaults."""
        material = self.model_dump(mode="json")
        material["runtime"]["options"] = self.runtime.resolved_options()
        return material

    @property
    def arm(self) -> str:
        """Return digest representing experimental condition without study paths."""
        return self._digests.arm

    @property
    def condition(self) -> str:
        """Return arm digest excluding plan attempt count."""
        return self._digests.condition

arm property

arm: str

Return digest representing experimental condition without study paths.

condition property

condition: str

Return arm digest excluding plan attempt count.

fingerprint property

fingerprint: str

Return deterministic configuration fingerprint.

declared staticmethod

declared(path: Path | str, section: str, key: str) -> bool

Check whether a setting key is explicitly declared in a TOML file.

Source code in src/reach/config.py

@staticmethod
def declared(path: Path | str, section: str, key: str) -> bool:
    """Check whether a setting key is explicitly declared in a TOML file."""
    payload = tomllib.loads(
        Path(path).expanduser().resolve().read_text(encoding="utf-8"),
    )
    return key in payload.get(section, {})

from_toml classmethod

from_toml(
    path: Path | str | None = None,
    *,
    skills: Path | str | None = None,
) -> RunConfig

Load and validate a RunConfig from a TOML configuration file.

Source code in src/reach/config.py

@classmethod
def from_toml(
    cls,
    path: Path | str | None = None,
    *,
    skills: Path | str | None = None,
) -> RunConfig:
    """Load and validate a RunConfig from a TOML configuration file."""
    resolved = (
        _discover_config_path(None)[0] if path is None else Path(path).expanduser().resolve()
    )
    if resolved is None:
        config = cls()
        if skills is not None:
            resolved_skills = expand_path(skills, base=Path.cwd())
            config = config.model_copy(
                update={"study": config.study.model_copy(update={"skills": resolved_skills})}
            )
        return config

    payload = tomllib.loads(resolved.read_text(encoding="utf-8"))
    study = dict(payload.get("study", {}))
    if skills is not None:
        study["skills"] = skills
    for key in ("skills", "queries", "workdir", "out"):
        raw = study.get(key)
        if raw is not None:
            study[key] = expand_path(raw, base=resolved.parent)
    payload["study"] = study
    return cls.model_validate(payload)

get_section

get_section(section_cls: type[T]) -> T

Retrieve the configuration section corresponding to section_cls.

Source code in src/reach/config.py

def get_section[T: BaseModel](self, section_cls: type[T]) -> T:
    """Retrieve the configuration section corresponding to section_cls."""
    attr_name = self._SECTION_MAP.get(section_cls)
    if attr_name is not None and hasattr(self, attr_name):
        val = getattr(self, attr_name)
        if isinstance(val, section_cls):
            return val
    for attr in self.__dict__:
        val = getattr(self, attr)
        if isinstance(val, section_cls):
            return val
    return section_cls()

material

material() -> dict[str, Any]

Dump configuration dictionary with agent options resolved to defaults.

Source code in src/reach/config.py

def material(self) -> dict[str, Any]:
    """Dump configuration dictionary with agent options resolved to defaults."""
    material = self.model_dump(mode="json")
    material["runtime"]["options"] = self.runtime.resolved_options()
    return material

require_queries

require_queries(hint: str = '') -> Path

Forward queries path requirement to study settings.

Source code in src/reach/config.py

def require_queries(self, hint: str = "") -> Path:
    """Forward queries path requirement to study settings."""
    return self.study.require_queries(hint)

require_skills

require_skills(hint: str = '') -> Path

Forward skills path requirement to study settings.

Source code in src/reach/config.py

def require_skills(self, hint: str = "") -> Path:
    """Forward skills path requirement to study settings."""
    return self.study.require_skills(hint)

require_workdir

require_workdir(hint: str = '') -> Path

Forward workdir path requirement to study settings.

Source code in src/reach/config.py

def require_workdir(self, hint: str = "") -> Path:
    """Forward workdir path requirement to study settings."""
    return self.study.require_workdir(hint)

resolve classmethod

resolve(
    section_cls: type[T],
    config: RunConfig | None = None,
    *,
    explicit_settings: T | None = None,
    **overrides: object,
) -> T

Resolve effective settings section from an optional RunConfig or defaults.

Source code in src/reach/config.py

@classmethod
def resolve[T: BaseModel](
    cls,
    section_cls: type[T],
    config: RunConfig | None = None,
    *,
    explicit_settings: T | None = None,
    **overrides: object,
) -> T:
    """Resolve effective settings section from an optional RunConfig or defaults."""
    section = config.get_section(section_cls) if config is not None else None
    active_overrides = dict(overrides)

    if issubclass(section_cls, RuntimeSettings):
        configured_runtime = section if isinstance(section, RuntimeSettings) else None
        base_runtime = (
            explicit_settings
            if isinstance(explicit_settings, RuntimeSettings)
            else (configured_runtime if configured_runtime is not None else RuntimeSettings())
        )
        resolved_runtime = base_runtime.with_overrides(**active_overrides)
        if config is not None:
            resolved_runtime = _propagate_registry_to_runtime(resolved_runtime, config.registry)
        return cast(T, resolved_runtime)

    resolved = resolve_sub_settings(
        section_cls,
        config_section=section,
        explicit_settings=explicit_settings,
        **active_overrides,
    )

    if isinstance(resolved, RegistrySettings):
        return cast(T, cls._resolve_registry_section(resolved, config))
    return cast(T, resolved)

resolve_section

resolve_section(
    section_cls: type[T],
    *,
    explicit_settings: T | None = None,
    **overrides: object,
) -> T

Resolve effective settings section against this RunConfig instance.

Source code in src/reach/config.py

def resolve_section[T: BaseModel](
    self,
    section_cls: type[T],
    *,
    explicit_settings: T | None = None,
    **overrides: object,
) -> T:
    """Resolve effective settings section against this RunConfig instance."""
    return self.resolve(
        section_cls,
        config=self,
        explicit_settings=explicit_settings,
        **overrides,
    )

with_overrides

with_overrides(**sections: dict[str, Any]) -> RunConfig

Return copy of configuration with section field overrides applied.

Source code in src/reach/config.py

def with_overrides(self, **sections: dict[str, Any]) -> RunConfig:
    """Return copy of configuration with section field overrides applied."""
    payload: dict[str, Any] = {}
    for section_name in type(self).model_fields:
        section_model = getattr(self, section_name)
        if isinstance(section_model, BaseModel):
            sec_dump = {
                **section_model.model_dump(exclude_defaults=True),
                **section_model.model_dump(exclude_unset=True),
            }
            if section_name == "runtime" and "agent" not in section_model.model_fields_set:
                sec_dump.pop("agent", None)
            if section_name == "optimize" and "workers" not in section_model.model_fields_set:
                sec_dump.pop("workers", None)
            if isinstance(section_model, LintSettings):
                if (
                    self.catalog.mode in {CatalogMode.NEIGHBORHOOD, CatalogMode.SINGLETON}
                    and section_model.catalog_budget_chars is None
                ):
                    sec_dump.pop("catalog_budget_chars", None)
                if (
                    "similarity_threshold" in self.retrieval.model_fields_set
                    and section_model.similarity_threshold
                    == self.retrieval.similarity_threshold
                ):
                    sec_dump.pop("similarity_threshold", None)
            payload[section_name] = sec_dump
        else:
            payload[section_name] = section_model

    for section, fields in sections.items():
        live = {k: v for k, v in (fields or {}).items() if v is not None}
        if section == "runtime":
            new_runtime = self.runtime.with_overrides(**live)
            sec_dump = {
                **new_runtime.model_dump(exclude_defaults=True),
                **new_runtime.model_dump(exclude_unset=True),
            }
            if "agent" not in new_runtime.model_fields_set:
                sec_dump.pop("agent", None)
            payload["runtime"] = sec_dump
        else:
            payload[section] = {**payload.get(section, {}), **live}
    return type(self).model_validate(payload)

RuntimeSettings

Bases: BaseModel

Configuration settings for agent runtime execution.

Source code in src/reach/config.py

class RuntimeSettings(BaseModel):
    """Configuration settings for agent runtime execution."""

    model_config = ConfigDict(frozen=True, extra="forbid")

    agent: str = Field(default_factory=default_agent)
    allowed_tools: tuple[str, ...] | None = None
    blocked_env_vars: tuple[str, ...] | None = None
    options: dict[str, object] = Field(default_factory=dict)
    timeout_s: int = Field(default=200, gt=0)
    max_turns: int = Field(default=3, ge=1)
    early_exit: bool = True

    @model_validator(mode="after")
    def _options_match_the_agent(self) -> Self:
        """Reject options the chosen agent does not accept, at load time."""
        self.resolved_options()
        return self

    def resolved_options(self) -> dict[str, Any]:
        """Return this agent's options with its own defaults filled in."""
        from reach.runtime import resolve_options

        resolved = resolve_options(self)
        return resolved.model_dump(mode="json") if resolved is not None else {}

    def with_overrides(self, **overrides: object) -> RuntimeSettings:
        """Return a copy with overrides applied, discarding stale options when the agent changes."""
        options_specified = "options" in overrides
        clean_overrides = {k: v for k, v in overrides.items() if v is not None}
        agent_specified = "agent" in clean_overrides
        target_agent = clean_overrides.get("agent", self.agent)
        same_agent = target_agent == self.agent
        model = clean_overrides.pop("model", None)

        if options_specified and overrides["options"] is None:
            base_opts: Any = {}
        else:
            base_opts = dict(self.options) if same_agent else {}
        options_override = clean_overrides.pop("options", None)
        if isinstance(options_override, Mapping):
            base_opts.update(options_override)
        elif options_override is not None:
            base_opts = options_override

        if model is not None and not isinstance(base_opts, dict):
            opts_type = type(base_opts).__name__
            msg = f"Cannot set 'model' override when options is of type {opts_type}"
            raise ValueError(msg)
        if model is not None:
            base_opts["model"] = model

        payload: dict[str, Any] = {
            **self.model_dump(exclude_unset=True),
            **clean_overrides,
            "options": base_opts,
        }
        if agent_specified or "agent" in self.model_fields_set:
            payload["agent"] = target_agent
        else:
            payload.pop("agent", None)
        return type(self).model_validate(payload)

    @classmethod
    def resolve_for_optimize(
        cls,
        config: Path | str | None = None,
        *,
        agent: str | None = None,
        options: Mapping[str, Any] | None = None,
    ) -> RuntimeSettings:
        """Resolve RuntimeSettings merging reach.toml [runtime.options] with overrides."""
        raw_cfg = load_config(config)
        raw_runtime = raw_cfg.get("runtime", {})
        cfg_agent = (
            str(raw_runtime.get("agent"))
            if isinstance(raw_runtime, Mapping) and raw_runtime.get("agent")
            else default_agent(config)
        )
        if isinstance(raw_runtime, Mapping) and isinstance(raw_runtime.get("options"), Mapping):
            raw_opts = dict(raw_runtime["options"])
            try:
                base_runtime = cls(agent=cfg_agent, options=raw_opts)
            except ValueError:
                base_runtime = cls(agent=cfg_agent)
        else:
            base_runtime = cls(agent=cfg_agent)
        overrides: dict[str, Any] = {}
        if agent is not None:
            overrides["agent"] = agent
        if options is not None:
            overrides["options"] = dict(options)
        return base_runtime.with_overrides(**overrides)

resolve_for_optimize classmethod

resolve_for_optimize(
    config: Path | str | None = None,
    *,
    agent: str | None = None,
    options: Mapping[str, Any] | None = None,
) -> RuntimeSettings

Resolve RuntimeSettings merging reach.toml [runtime.options] with overrides.

Source code in src/reach/config.py

@classmethod
def resolve_for_optimize(
    cls,
    config: Path | str | None = None,
    *,
    agent: str | None = None,
    options: Mapping[str, Any] | None = None,
) -> RuntimeSettings:
    """Resolve RuntimeSettings merging reach.toml [runtime.options] with overrides."""
    raw_cfg = load_config(config)
    raw_runtime = raw_cfg.get("runtime", {})
    cfg_agent = (
        str(raw_runtime.get("agent"))
        if isinstance(raw_runtime, Mapping) and raw_runtime.get("agent")
        else default_agent(config)
    )
    if isinstance(raw_runtime, Mapping) and isinstance(raw_runtime.get("options"), Mapping):
        raw_opts = dict(raw_runtime["options"])
        try:
            base_runtime = cls(agent=cfg_agent, options=raw_opts)
        except ValueError:
            base_runtime = cls(agent=cfg_agent)
    else:
        base_runtime = cls(agent=cfg_agent)
    overrides: dict[str, Any] = {}
    if agent is not None:
        overrides["agent"] = agent
    if options is not None:
        overrides["options"] = dict(options)
    return base_runtime.with_overrides(**overrides)

resolved_options

resolved_options() -> dict[str, Any]

Return this agent's options with its own defaults filled in.

Source code in src/reach/config.py

def resolved_options(self) -> dict[str, Any]:
    """Return this agent's options with its own defaults filled in."""
    from reach.runtime import resolve_options

    resolved = resolve_options(self)
    return resolved.model_dump(mode="json") if resolved is not None else {}

with_overrides

with_overrides(**overrides: object) -> RuntimeSettings

Return a copy with overrides applied, discarding stale options when the agent changes.

Source code in src/reach/config.py

def with_overrides(self, **overrides: object) -> RuntimeSettings:
    """Return a copy with overrides applied, discarding stale options when the agent changes."""
    options_specified = "options" in overrides
    clean_overrides = {k: v for k, v in overrides.items() if v is not None}
    agent_specified = "agent" in clean_overrides
    target_agent = clean_overrides.get("agent", self.agent)
    same_agent = target_agent == self.agent
    model = clean_overrides.pop("model", None)

    if options_specified and overrides["options"] is None:
        base_opts: Any = {}
    else:
        base_opts = dict(self.options) if same_agent else {}
    options_override = clean_overrides.pop("options", None)
    if isinstance(options_override, Mapping):
        base_opts.update(options_override)
    elif options_override is not None:
        base_opts = options_override

    if model is not None and not isinstance(base_opts, dict):
        opts_type = type(base_opts).__name__
        msg = f"Cannot set 'model' override when options is of type {opts_type}"
        raise ValueError(msg)
    if model is not None:
        base_opts["model"] = model

    payload: dict[str, Any] = {
        **self.model_dump(exclude_unset=True),
        **clean_overrides,
        "options": base_opts,
    }
    if agent_specified or "agent" in self.model_fields_set:
        payload["agent"] = target_agent
    else:
        payload.pop("agent", None)
    return type(self).model_validate(payload)

Severity

Bases: StrEnum

Specify the severity level for a lint diagnostic.

Source code in src/reach/config.py

class Severity(StrEnum):
    """Specify the severity level for a lint diagnostic."""

    ERROR = "error"
    IGNORE = "ignore"
    INFO = "info"
    WARN = "warn"

StudySettings

Bases: BaseModel

Configuration settings for experiment inputs, workspaces, and outputs.

Source code in src/reach/config.py

class StudySettings(BaseModel):
    """Configuration settings for experiment inputs, workspaces, and outputs."""

    model_config = ConfigDict(frozen=True, extra="forbid")

    auto_queries: bool = True
    skills: Path | None = None
    queries: Path | None = None
    workdir: Path | None = None
    out: Path | None = None
    tag: str = ""
    partial: bool = False
    catalog: str | None = "auto"
    rescope: bool = False
    scales: tuple[int, ...] | None = None
    anchor: int | tuple[str, ...] | str | None = None
    trusted: bool = False
    bootstrap_iterations: int = Field(default=200, ge=10)
    bootstrap_seed: int | None = Field(default=42)
    catalog_replicates: Annotated[int, Field(ge=1, le=20)] = 1

    @field_validator("anchor", mode="before")
    @classmethod
    def _coerce_anchor(cls, value: object) -> int | tuple[str, ...] | str | None:
        """Coerce anchor specification into integer count, skill names tuple, or 'all'."""
        if value is None:
            return None
        if isinstance(value, int):
            if value <= 0:
                msg = f"anchor size must be a positive integer, got {value}"
                raise ValueError(msg)
            return value
        raw_items: Sequence[object]
        if isinstance(value, str):
            val = value.strip()
            if not val:
                return None
            if val.lower() == "all":
                return "all"
            if val.isdigit():
                iv = int(val)
                if iv <= 0:
                    msg = f"anchor size must be a positive integer, got {iv}"
                    raise ValueError(msg)
                return iv
            raw_items = val.split(",")
        elif isinstance(value, (list, tuple)):
            raw_items = value
        else:
            msg = f"invalid anchor specification: {value}"
            raise ValueError(msg)
        return tuple(str(p).strip() for p in raw_items if str(p).strip())

    @field_validator("scales", mode="before")
    @classmethod
    def _coerce_scales(cls, value: object) -> tuple[int, ...] | None:
        """Coerce comma-separated string or sequence into tuple of positive integers."""
        if value is None:
            return None
        if isinstance(value, str):
            parts = [int(p.strip()) for p in value.split(",") if p.strip()]
            return tuple(parts)
        if isinstance(value, (list, tuple)):
            return tuple(int(v) for v in value)
        msg = f"invalid scales specification: {value}"
        raise ValueError(msg)

    @field_validator("skills", "queries", "workdir", "out", mode="before")
    @classmethod
    def _expand(cls, value: object) -> object:
        """Expand path variables and user directories for filesystem paths."""
        if value is None:
            return None
        return str(expand_path(value))

    def require_queries(self, hint: str = "") -> Path:
        """Return the configured queries path or raise a ValueError."""
        if self.queries is None:
            msg = f"queries path is required{f': {hint}' if hint else ''}"
            raise ValueError(msg)
        return self.queries

    def require_skills(self, hint: str = "") -> Path:
        """Return the configured skills path or raise a ValueError."""
        if self.skills is None:
            msg = f"skills path is required{f': {hint}' if hint else ''}"
            raise ValueError(msg)
        return self.skills

    def require_workdir(self, hint: str = "") -> Path:
        """Return the resolved configured workdir path or raise a ValueError."""
        if self.workdir is None:
            msg = f"workdir path is required{f': {hint}' if hint else ''}"
            raise ValueError(msg)
        return self.workdir.resolve()

require_queries

require_queries(hint: str = '') -> Path

Return the configured queries path or raise a ValueError.

Source code in src/reach/config.py

def require_queries(self, hint: str = "") -> Path:
    """Return the configured queries path or raise a ValueError."""
    if self.queries is None:
        msg = f"queries path is required{f': {hint}' if hint else ''}"
        raise ValueError(msg)
    return self.queries

require_skills

require_skills(hint: str = '') -> Path

Return the configured skills path or raise a ValueError.

Source code in src/reach/config.py

def require_skills(self, hint: str = "") -> Path:
    """Return the configured skills path or raise a ValueError."""
    if self.skills is None:
        msg = f"skills path is required{f': {hint}' if hint else ''}"
        raise ValueError(msg)
    return self.skills

require_workdir

require_workdir(hint: str = '') -> Path

Return the resolved configured workdir path or raise a ValueError.

Source code in src/reach/config.py

def require_workdir(self, hint: str = "") -> Path:
    """Return the resolved configured workdir path or raise a ValueError."""
    if self.workdir is None:
        msg = f"workdir path is required{f': {hint}' if hint else ''}"
        raise ValueError(msg)
    return self.workdir.resolve()

load_config

load_config(
    config_path: Path | str | None = None,
) -> dict[str, Any]

Load configuration from bundled reach.toml overlaid with optional project config.

Source code in src/reach/config.py

def load_config(config_path: Path | str | None = None) -> dict[str, Any]:
    """Load configuration from bundled reach.toml overlaid with optional project config."""
    base: dict[str, Any] = {}
    if BUNDLED_CONFIG_PATH.is_file():
        base = tomllib.loads(BUNDLED_CONFIG_PATH.read_text(encoding="utf-8"))

    target, _ = _discover_config_path(config_path)
    if target is not None:
        override = tomllib.loads(target.read_text(encoding="utf-8"))
        base = _deep_merge(base, override)

    return base