agentic_core/config.py

"""Environment-based configuration for the agentic core.

Secrets are only ever read from environment variables / `.env` and are never
logged. The effective API key prefers an explicit ``LLM_API_KEY`` and falls
back to ``CURSOR_API_KEY`` so the provider can be swapped later.
"""

from __future__ import annotations

from functools import lru_cache
from pathlib import Path

from pydantic_settings import BaseSettings, SettingsConfigDict

ROOT_DIR = Path(__file__).resolve().parent.parent


class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=ROOT_DIR / ".env",
        env_file_encoding="utf-8",
        extra="ignore",
    )

    # Credentials
    cursor_api_key: str = ""
    kimi_api_key: str = ""
    openrouter_api_key: str = ""
    groq_api_key: str = ""
    gemini_api_key: str = ""
    llm_api_key: str = ""

    # LLM provider selection
    llm_provider: str = "cursor"
    llm_model: str = "default"
    # Google model routed through Cursor Cloud Agents. Used by the artifact
    # summarizer (when enabled) and as the Cursor provider's default when
    # LLM_MODEL is unset.
    llm_fast_model: str = "gemini-3.7-flash"
    # Ask the Cursor Cloud Agents API to run composer models in fast mode.
    cursor_fast_mode: bool = True
    # When enabled, the orchestrator spends an LLM call (fastest model) to
    # summarize each artifact before it is handed downstream. Default OFF:
    # deterministic Python digests are sufficient for cross-artifact contracts
    # and cost a full LLM call per artifact (~60s + provider tokens each).
    summarize_with_llm: bool = False
    llm_base_url: str = "https://api.cursor.com/v1"
    kimi_base_url: str = "https://api.moonshot.cn/v1"
    openrouter_base_url: str = "https://openrouter.ai/api/v1"
    groq_base_url: str = "https://api.groq.com/openai/v1"
    gemini_base_url: str = "https://generativelanguage.googleapis.com/v1beta/openai"
    llm_request_timeout_s: float = 120.0
    # Must clear the largest artifact the model is asked to emit. At 3072 the
    # database, api, devops and requirements responses were all cut off
    # mid-object on the OpenAI-compatible providers.
    llm_max_tokens: int = 8192
    llm_poll_interval_s: float = 0.5
    llm_poll_timeout_s: float = 300.0
    # Fraction of llm_poll_timeout_s after which a repair attempt is skipped
    # (fail fast so a retry of the full agent runs sooner). 0.55 = 165s of 300s.
    llm_repair_skip_fraction: float = 0.55

    # Workflow behaviour
    structured_output_max_retries: int = 1
    # The reviewer runs at most this many times per workflow (exactly one
    # review round: PASS or regenerate once, then complete).
    max_review_rounds: int = 1
    # Each artifact may be regenerated at most this many times per workflow.
    max_artifact_revisions: int = 1
    # Bounded retries for transient provider/transport failures.
    max_llm_retries: int = 1

    # Persistence
    data_dir: Path = ROOT_DIR / "data"

    @property
    def projects_dir(self) -> Path:
        return self.data_dir / "projects"

    @property
    def runs_dir(self) -> Path:
        return self.data_dir / "runs"

    @property
    def artifacts_dir(self) -> Path:
        return self.data_dir / "artifacts"

    @property
    def db_path(self) -> Path:
        return self.data_dir / "b2d.db"

    def effective_provider(self) -> str:
        return (self.llm_provider or "cursor").strip().lower()

    def effective_model(self) -> str:
        if self.effective_provider() == "kimi":
            return (
                self.llm_model
                if self.llm_model and self.llm_model != "default"
                else "moonshotai/Kimi-K2-Instruct"
            )
        if self.effective_provider() == "groq":
            return (
                self.llm_model
                if self.llm_model and self.llm_model != "default"
                else "openai/gpt-oss-120b"
            )
        if self.effective_provider() == "gemini":
            return (
                self.llm_model
                if self.llm_model and self.llm_model != "default"
                else "gemini-3.7-flash"
            )
        if self.effective_provider() == "openrouter":
            return (
                self.llm_model
                if self.llm_model and self.llm_model != "default"
                else "nvidia/nemotron-3-ultra-550b-a55b:free"
            )
        if self.llm_model and self.llm_model != "default":
            return self.llm_model
        # The Cursor Cloud Agents default is Cursor's fastest model.
        return self.llm_fast_model

    def effective_model_name(self) -> str:
        return self.effective_model()

    def effective_base_url(self) -> str:
        if self.effective_provider() == "kimi":
            return self.kimi_base_url or self.llm_base_url or "https://api.moonshot.cn/v1"
        if self.effective_provider() == "groq":
            return self.groq_base_url or "https://api.groq.com/openai/v1"
        if self.effective_provider() == "gemini":
            return self.gemini_base_url or "https://generativelanguage.googleapis.com/v1beta/openai"
        if self.effective_provider() == "openrouter":
            return self.openrouter_base_url or self.llm_base_url or "https://openrouter.ai/api/v1"
        return self.llm_base_url or "https://api.cursor.com/v1"

    def effective_base_url_value(self) -> str:
        return self.effective_base_url()

    def effective_api_key(self) -> str:
        if self.effective_provider() == "groq":
            return self.groq_api_key or self.llm_api_key or ""
        if self.effective_provider() == "gemini":
            return self.gemini_api_key or self.llm_api_key or ""
        if self.effective_provider() == "kimi":
            return self.kimi_api_key or self.llm_api_key or ""
        if self.effective_provider() == "openrouter":
            return self.openrouter_api_key or self.llm_api_key or ""
        return self.llm_api_key or self.cursor_api_key or ""

    def effective_api_key_value(self) -> str:
        return self.effective_api_key()

    def ensure_dirs(self) -> None:
        # data_dir is the parent of the SQLite database; runs/artifacts remain
        # directory-based, so keep creating them.
        for path in (self.data_dir, self.runs_dir, self.artifacts_dir):
            path.mkdir(parents=True, exist_ok=True)

    def check_credentials(self) -> None:
        if self.effective_provider() == "groq":
            if not self.effective_api_key_value():
                raise RuntimeError("No Groq API key configured. Set GROQ_API_KEY or LLM_API_KEY in `.env`.")
            return
        if self.effective_provider() == "gemini":
            if not self.effective_api_key_value():
                raise RuntimeError("No Gemini API key configured. Set GEMINI_API_KEY or LLM_API_KEY in `.env`.")
            return
        if self.effective_provider() == "kimi":
            if not self.effective_api_key_value():
                raise RuntimeError(
                    "No Kimi API key configured. Set KIMI_API_KEY or LLM_API_KEY in `.env`."
                )
            return
        if self.effective_provider() == "openrouter":
            if not self.effective_api_key_value():
                raise RuntimeError(
                    "No OpenRouter API key configured. Set OPENROUTER_API_KEY or LLM_API_KEY in `.env`."
                )
            return
        if not self.effective_api_key_value():
            raise RuntimeError(
                "No Cursor API key configured. Set CURSOR_API_KEY or LLM_API_KEY in `.env`."
            )


@lru_cache
def get_settings() -> Settings:
    settings = Settings()
    settings.ensure_dirs()
    settings.check_credentials()
    return settings