control_plane/settings.py

"""Runtime configuration for the control plane.

Settings are managed with ``pydantic-settings`` (``BaseSettings``), which reads
from environment variables / a Hugging Face Space secret and an optional local
``.env`` file. See:
https://pydantic.dev/docs/validation/latest/concepts/pydantic_settings/

The model API key is *never* required at import time: a missing key must not
crash the app at startup, so it defaults to ``None`` and is only needed at the
moment a live model is built. All settings share the ``OPENROUTER_``
env prefix, so the fields ``model``, ``api_key`` and ``max_turns`` map to
``OPENROUTER_MODEL``, ``OPENROUTER_API_KEY`` and ``OPENROUTER_MAX_TURNS``.
"""

from __future__ import annotations

from pydantic import Field, field_validator
from pydantic_settings import BaseSettings, SettingsConfigDict

# Default model id. Overridable via OPENROUTER_MODEL with no
# code changes.
DEFAULT_MODEL = "openai/gpt-5-mini"

# Default cap on how many turns one governed-loop run may take. See the
# ``max_turns`` field below for what this protects against.
DEFAULT_MAX_TURNS = 10

# Default cap on how many tool calls a live run may make in total. Each tool call
# is one model round-trip (~10–15s), so this is the real bound on a live run's
# wall-clock time (``max_turns`` only counts structured turns, while a model can
# chain many tool calls inside a single turn). See ``tool_call_budget`` below.
DEFAULT_TOOL_CALL_BUDGET = 6


class Settings(BaseSettings):
    """Immutable snapshot of runtime configuration.

    ``api_key`` may be ``None``. Constructing this never raises on a missing
    key; callers that actually reach the model handle its absence.
    """

    model_config = SettingsConfigDict(
        env_prefix="OPENROUTER_",
        env_file=".env",
        env_file_encoding="utf-8",
        extra="ignore",
        frozen=True,
    )

    model: str = DEFAULT_MODEL
    api_key: str | None = None

    # Hard ceiling on the number of turns a single governed-loop run may take
    # (one turn = propose → gate → maybe execute → feed result back). The agent
    # normally stops on its own by emitting a final answer; this is the safety
    # net for when it does not — e.g. it loops, keeps retrying a failing action,
    # or never concludes. Without it a live run could spin indefinitely, burning
    # model calls. ``ge=1`` rejects a non-positive cap (which would do nothing).
    max_turns: int = Field(default=DEFAULT_MAX_TURNS, ge=1)

    # Hard ceiling on the TOTAL number of tool calls a live run may make. This is
    # the real bound on a live run's wall-clock time: each tool call is a slow
    # model round-trip, and a model can chain many within a single turn (which
    # ``max_turns`` does not bound). Lower it for a snappier demo, raise it to let
    # the agent do more. ``ge=1`` rejects a non-positive cap.
    tool_call_budget: int = Field(default=DEFAULT_TOOL_CALL_BUDGET, ge=1)

    @field_validator("model", mode="before")
    @classmethod
    def _model_default_when_blank(cls, v: object) -> object:
        """An unset or whitespace-only model id falls back to the default."""
        if v is None or (isinstance(v, str) and not v.strip()):
            return DEFAULT_MODEL
        return v

    @field_validator("max_turns", mode="before")
    @classmethod
    def _max_turns_default_when_blank(cls, v: object) -> object:
        """An unset or whitespace-only value falls back to the default."""
        if v is None or (isinstance(v, str) and not v.strip()):
            return DEFAULT_MAX_TURNS
        return v

    @field_validator("tool_call_budget", mode="before")
    @classmethod
    def _tool_budget_default_when_blank(cls, v: object) -> object:
        """An unset or whitespace-only value falls back to the default."""
        if v is None or (isinstance(v, str) and not v.strip()):
            return DEFAULT_TOOL_CALL_BUDGET
        return v

    @field_validator("api_key", mode="before")
    @classmethod
    def _blank_key_is_none(cls, v: object) -> object:
        """Treat an empty/whitespace key as absent."""
        if isinstance(v, str) and not v.strip():
            return None
        return v

    @property
    def has_api_key(self) -> bool:
        return bool(self.api_key)


def load_settings(env: dict[str, str] | None = None) -> Settings:
    """Load settings from the environment (and ``.env``).

    Parameters
    ----------
    env:
        Optional mapping of ``OPENROUTER_*`` variables to read from instead of
        the process environment (used by tests). When provided, both fields are
        passed explicitly so the result is isolated from ``os.environ`` and any
        ``.env`` file, keeping tests deterministic.
    """
    if env is None:
        return Settings()
    return Settings(
        _env_file=None,
        model=env.get("OPENROUTER_MODEL", DEFAULT_MODEL),
        api_key=env.get("OPENROUTER_API_KEY"),
        max_turns=env.get("OPENROUTER_MAX_TURNS", DEFAULT_MAX_TURNS),
        tool_call_budget=env.get("OPENROUTER_TOOL_CALL_BUDGET", DEFAULT_TOOL_CALL_BUDGET),
    )