orchestrata_ultimate.core.infra_config

Reference › orchestrata_ultimate › core › infra_config

The typed infrastructure configuration for the whole system, built on Pydantic Settings and described in the core narrative. InfraConfig holds the settings for Redis, the graph endpoint, the per-provider LLM settings (DeepSeek, Anthropic, Kimi, Ollama), the default provider and model, the workspace and Prolog directories, and the swarm switches. Its constructor accepts an explicit environment mapping, validators coerce the port, swarm flag and retry count, and _env_file_if_readable drops a .env file the process cannot read instead of failing on it. A module-level instance feeds a set of upper-case constants (REDIS_HOST, GRAPH_URI, LLM_PROVIDER and others) that older code imports; _resolve_bin locates the claude and aider binaries from an environment variable, PATH, or fallback directories.

Per-volk model choice lives in VOLK_CONFIG, built by _build_volk_config from a base table of VolkConfig provider and model pairs. When ORCHESTRATA_LLM_PROVIDER is set, _apply_provider_override switches every volk to that provider, using ORCHESTRATA_LLM_MODEL and ORCHESTRATA_LLM_REASONING_MODEL (or the provider table) and giving the reasoning model to the volks in the reasoning set. get_volk_provider and get_volk_model read the table. validate returns warnings for an empty key of a provider in use or an empty Redis host. core.config.infra re-exports this module.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

VolkConfig (pydantic)

Bases: BaseSettings · Line 70

Per-volk LLM configuration.

InfraConfig (pydantic)

Bases: BaseSettings · Line 77

Global infrastructure configuration.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/core/infra_config.py

"""Infrastructure configuration using Pydantic Settings.

SPEC: Global infrastructure configuration with environment variable
fallbacks, validation, and type safety via Pydantic BaseSettings.
"""

import os
import shutil
from collections.abc import Mapping
from pathlib import Path
from typing import Any

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

from orchestrata_ultimate.core.base.traces import traces


def _env_file_if_readable(path: str) -> str | None:
    """``path`` for pydantic-settings' ``env_file``, or ``None`` when this
    process cannot READ it.

    A third independent direct-``.env``-reader alongside systemd's
    ``EnvironmentFile=`` and ``core.env_bootstrap.load_env_contract()`` —
    pydantic-settings' own dotenv provider (``dotenv.main.dotenv_values`` ->
    ``open()``) raises an uncaught ``PermissionError`` rather than treating
    "exists but unreadable" as "absent" the way a missing file already is.
    Measured 2026-08-13: the containerized TEST kernel/gateway bind-mount the
    repo read-only into an unprivileged Incus container, which UID-shifts
    the SOPS+age-hardened 0600 ``.env`` to an owner container-root cannot
    read; systemd's ``EnvironmentFile=`` (a container-local, correctly-owned
    copy) already populated ``os.environ`` before ``InfraConfig`` is ever
    instantiated, so skipping the redundant direct read here loses nothing.

    ``os.access``, not ``Path.exists()`` — ``exists()`` only needs execute on
    the parent directory (stat), which an 0600 file still grants; only an
    actual read attempt (or ``os.access(..., os.R_OK)``) sees the denial.
    """
    return path if os.access(path, os.R_OK) else None


def _resolve_bin(
    env_var: str,
    name: str,
    fallback_dirs: list[str] | None = None,
    env: Mapping[str, str] | None = None,
) -> str:
    """Resolve a binary path from env var, PATH, or common fallback dirs.

    ``env`` defaults to ``os.environ`` and is exposed so tests can exercise
    the resolver without mutating the process environment.
    """
    env = env if env is not None else os.environ
    path = env.get(env_var, "")
    if path and os.path.isfile(path):
        return path
    found = shutil.which(name)
    if found:
        return found
    for d in fallback_dirs or [
        str(Path.home() / ".local" / "bin"),
        "/usr/local/bin",
    ]:
        candidate = os.path.join(d, name)
        if os.path.isfile(candidate):
            return candidate
    return name


class VolkConfig(BaseSettings):
    """Per-volk LLM configuration."""

    provider: str = "claude"
    model: str = "sonnet"


class InfraConfig(BaseSettings):
    """Global infrastructure configuration.

    All fields load from environment variables with sensible defaults.
    An explicit ``env`` mapping may be supplied so callers (and tests) can
    construct a config without mutating ``os.environ``.
    """

    model_config = SettingsConfigDict(
        env_prefix="ORCHESTRATA_",
        env_file=_env_file_if_readable(".env"),
        env_file_encoding="utf-8",
        extra="ignore",
        populate_by_name=True,
    )

    def __init__(self, env: Mapping[str, str] | None = None, **kwargs: Any) -> None:
        """Build settings from an optional explicit env mapping.

        When ``env`` is provided, values from that mapping take precedence over
        ``os.environ`` for fields that are present; missing values fall back to
        the normal pydantic-settings resolution path.

        ``**kwargs`` is typed ``Any`` (not ``object``) because it is forwarded
        verbatim to ``BaseSettings.__init__``, whose signature accepts both
        model field values and pydantic-settings' own typed control kwargs
        (``_case_sensitive``, ``_env_prefix``, ...); ``object`` values are not
        assignable to those narrower parameter types even though at runtime
        we never pass them.
        """
        if env is not None:
            merged: dict[str, Any] = {}
            prefix = InfraConfig.model_config.get("env_prefix", "")
            for field_name, field_info in InfraConfig.model_fields.items():
                env_key = (
                    field_info.alias
                    if field_info.alias
                    else f"{prefix}{field_name.upper()}"
                )
                if env_key in env:
                    merged[field_name] = env[env_key]
            kwargs = {**merged, **kwargs}
        super().__init__(**kwargs)

    # ------------------------------------------------------------------
    # Redis
    # ------------------------------------------------------------------
    redis_host: str = Field(default="localhost", alias="REDIS_HOST")
    redis_port: int = Field(default=6379, alias="REDIS_PORT")

    # ------------------------------------------------------------------
    # Operational knowledge graph (orchgraph cluster, `orchkg` database)
    # ------------------------------------------------------------------
    # Sourced from KG_ROUTING_URI, not NEO4J_URI: the NEO4J_* env keys were
    # retired with the Neo4j containers on 2026-07-27 (migration plan Step
    # 5.1). The ATTRIBUTE is graph_* (renamed from neo4j_* in the graph-
    # rename wave, F10) — project-owned vocabulary, distinct from the
    # `neo4j` driver library import itself (kept as-is; it's genuinely what
    # connects to orchgraph) and from the env key, which was already wrong.
    #
    # The old default was `bolt://localhost:7687`. Nothing in this
    # infrastructure has ever listened on localhost (every graph is an Incus
    # container on 10.53.1.x), so that default could only ever turn a config
    # gap into a connection-refused that reads like an outage. Empty means
    # unset, and callers route through core.graph_factory.resolve_uri(), which
    # raises with the key name instead of dialling a dead address.
    graph_uri: str = Field(default="", alias="KG_ROUTING_URI")
    # No credentials: every remaining graph endpoint negotiates no auth.
    # Retained (empty) so the attribute surface stays stable for the config's
    # consumers; see core.graph_factory._resolve_auth.
    graph_username: str = ""
    graph_password: str = ""

    # ------------------------------------------------------------------
    # LLM Provider API Keys
    # ------------------------------------------------------------------
    deepseek_api_key: str = ""
    deepseek_base_url: str = "https://api.deepseek.com/v1"
    deepseek_model: str = "deepseek-v4-flash"
    deepseek_reasoner_model: str = "deepseek-v4-pro"

    anthropic_api_key: str = ""

    kimi_api_key: str = ""
    kimi_base_url: str = "https://api.kimi.com/coding/v1"
    kimi_model: str = "kimi-for-coding"

    ollama_base_url: str = ""
    ollama_model: str = "qwen3:1.7b"

    # ------------------------------------------------------------------
    # LLM Defaults
    # ------------------------------------------------------------------
    llm_provider: str = "claude"
    llm_model: str = "sonnet"
    llm_reasoning_model: str = "opus"

    # ------------------------------------------------------------------
    # Workspace & Paths
    # ------------------------------------------------------------------
    workspace: str = Field(default_factory=lambda: str(Path.home() / "workspace"))
    prolog_dir: str = Field(
        default_factory=lambda: str(Path(__file__).resolve().parent.parent / "prolog")
    )

    # ------------------------------------------------------------------
    # Swarm
    # ------------------------------------------------------------------
    swarm_enabled: bool = True
    swarm_retries: int = 2

    # ------------------------------------------------------------------
    # ACP / Junior Drivers
    # ------------------------------------------------------------------
    aider_model: str = ""

    @field_validator("redis_port", mode="before")
    @classmethod
    def _coerce_redis_port(cls, v: object) -> int:
        return int(v) if v is not None else 6379  # type: ignore[call-overload]

    @field_validator("swarm_enabled", mode="before")
    @classmethod
    def _coerce_swarm_enabled(cls, v: object) -> bool:
        if isinstance(v, str):
            return v.lower() in ("1", "true", "yes", "on")
        return bool(v)

    @field_validator("swarm_retries", mode="before")
    @classmethod
    def _coerce_swarm_retries(cls, v: object) -> int:
        return int(v) if v is not None else 2  # type: ignore[call-overload]


# ------------------------------------------------------------------------------
# Module-level re-exports for backwards compatibility
# ------------------------------------------------------------------------------

_cfg = InfraConfig()

REDIS_HOST: str = _cfg.redis_host
REDIS_PORT: int = _cfg.redis_port
GRAPH_URI: str = _cfg.graph_uri
GRAPH_USER: str = _cfg.graph_username
GRAPH_PASSWORD: str = _cfg.graph_password
PROLOG_RULES_DIR: str = _cfg.prolog_dir
WORKSPACE_DIR: str = _cfg.workspace
PROJECT_WORKSPACE_ROOT: str = _cfg.workspace

LLM_PROVIDER: str = _cfg.llm_provider
LLM_MODEL: str = _cfg.llm_model

DEEPSEEK_API_KEY: str = _cfg.deepseek_api_key
DEEPSEEK_BASE_URL: str = _cfg.deepseek_base_url
DEEPSEEK_MODEL: str = _cfg.deepseek_model
DEEPSEEK_REASONER_MODEL: str = _cfg.deepseek_reasoner_model

ANTHROPIC_API_KEY: str = _cfg.anthropic_api_key

KIMI_API_KEY: str = _cfg.kimi_api_key
KIMI_BASE_URL: str = _cfg.kimi_base_url
KIMI_MODEL: str = _cfg.kimi_model

OLLAMA_BASE_URL: str = _cfg.ollama_base_url
OLLAMA_MODEL: str = _cfg.ollama_model

SWARM_ENABLED: bool = _cfg.swarm_enabled
SWARM_MAX_RETRIES: int = _cfg.swarm_retries

CLAUDE_CODE_PATH: str = _resolve_bin("CLAUDE_CODE_PATH", "claude")
AIDER_PATH: str = _resolve_bin("AIDER_PATH", "aider")
AIDER_MODEL: str = _cfg.aider_model

# ------------------------------------------------------------------------------
# Per-volk configuration
# ------------------------------------------------------------------------------
_VOLK_CONFIG_BASE: dict[str, dict[str, str]] = {
    "U:Cx": {"provider": "claude", "model": "sonnet"},
    "U:BA": {"provider": "claude", "model": "sonnet"},
    "U:PM": {"provider": "claude", "model": "sonnet"},
    "U:Ux": {"provider": "claude", "model": "sonnet"},
    "U:Ar": {"provider": "claude", "model": "opus"},
    "U:Cd": {"provider": "claude", "model": "sonnet"},
    "J:Junior": {"provider": "claude", "model": "sonnet"},
    "U:Qa": {"provider": "claude", "model": "sonnet"},
    "U:Cp": {"provider": "claude", "model": "sonnet"},
}

# G195: env var override — apply to all volks when set
_PROVIDER_MODELS: dict[str, dict[str, str]] = {
    "claude": {"standard": "sonnet", "reasoning": "opus"},
    "deepseek": {"standard": "deepseek-v4-flash", "reasoning": "deepseek-v4-pro"},
    "kimi": {"standard": "kimi-for-coding", "reasoning": "kimi-for-coding"},
    "qwen": {"standard": "qwen3.6-plus", "reasoning": "qwen3.6-plus"},
    "ollama": {"standard": "qwen3:1.7b", "reasoning": "qwen3:1.7b"},
    "anthropic": {
        # Kept in sync with the Anthropic adapter registration in
        # llm/client.py:_register_anthropic (was stale claude-*-4-5-20250514).
        "standard": "claude-sonnet-4-6",
        "reasoning": "claude-opus-4-6",
    },
}
_REASONING_VOLKS: set[str] = {"U:Ar"}


def _apply_provider_override(
    env: Mapping[str, str],
    volk_config: dict[str, dict[str, str]],
) -> None:
    """Mutate ``volk_config`` in place according to ``ORCHESTRATA_LLM_PROVIDER``.

    The override logic mirrors the historical G195 module-load-time block, but
    is exposed as a pure function so tests can exercise it without reloading
    the module or mutating ``os.environ``.
    """
    provider = env.get("ORCHESTRATA_LLM_PROVIDER", "")
    if not provider:
        return
    models = _PROVIDER_MODELS.get(provider, {})
    std = env.get("ORCHESTRATA_LLM_MODEL", models.get("standard", ""))
    rsn = env.get(
        "ORCHESTRATA_LLM_REASONING_MODEL",
        models.get("reasoning", std),
    )
    for key in volk_config:
        volk_config[key]["provider"] = provider
        volk_config[key]["model"] = rsn if key in _REASONING_VOLKS else std


def _build_volk_config(
    env: Mapping[str, str] | None = None,
) -> dict[str, dict[str, str]]:
    """Return a fresh volk config table derived from ``env``.

    Defaults to ``os.environ`` so the module-level runtime table remains
    env-aware while individual tests can supply explicit mappings.
    """
    env = env if env is not None else os.environ
    volk_config = {k: dict(v) for k, v in _VOLK_CONFIG_BASE.items()}
    _apply_provider_override(env, volk_config)
    return volk_config


# Module-level runtime table: env-aware copy of the static base table.
VOLK_CONFIG: dict[str, dict[str, str]] = _build_volk_config(os.environ)

JUNIOR_VOLK: str = VOLK_CONFIG["J:Junior"]["provider"]


@traces(spec="028", reqs=["REQ-028-022", "REQ-028-026"])
def get_volk_provider(volk: str) -> str:
    """Get the LLM provider for a specific volk."""
    cfg = VOLK_CONFIG.get(volk, {})
    return cfg.get("provider", LLM_PROVIDER)


@traces(spec="028", reqs=["REQ-028-023"])
def get_volk_model(volk: str) -> str:
    """Get the LLM model for a specific volk."""
    cfg = VOLK_CONFIG.get(volk, {})
    return cfg.get("model", LLM_MODEL)


def validate(env: Mapping[str, str] | None = None) -> list[str]:
    """Validate critical config at boot time. Returns list of warnings.

    ``env`` defaults to ``os.environ``; callers may pass an explicit mapping to
    validate a different environment without mutating the process env.
    """
    env = env if env is not None else os.environ
    cfg = InfraConfig(env=env)
    warnings: list[str] = []
    volk_config = _build_volk_config(env)
    providers_used = {v.get("provider", cfg.llm_provider) for v in volk_config.values()}
    if "deepseek" in providers_used and not cfg.deepseek_api_key:
        warnings.append("DEEPSEEK_API_KEY is empty — DeepSeek calls will fail")
    if "anthropic" in providers_used and not cfg.anthropic_api_key:
        warnings.append("ANTHROPIC_API_KEY is empty — Anthropic calls will fail")
    if "kimi" in providers_used and not cfg.kimi_api_key:
        warnings.append("KIMI_API_KEY is empty — Kimi calls will fail")
    if not cfg.redis_host:
        warnings.append("REDIS_HOST is empty — set it in .env or environment")
    return warnings