orchestrata_ultimate.harness.config

Reference › orchestrata_ultimate › harness › config

Builds a HarnessConfig from three layers, highest priority first: environment variables, the YAML file (harness/config.yaml by default), then the dataclass defaults in models. load_config(path, environ) starts from the defaults, overlays the YAML through _apply_yaml when the file exists (it logs a warning and keeps defaults when it does not), overlays the environment through _apply_env, and creates the snapshot, history and diagnostics directories. _apply_env maps HARNESS_DEFAULT_LLM, HARNESS_REASONING_LLM, REDIS_HOST, REDIS_PORT and KG_ROUTING_URI to the matching fields, and ignores a value that is unset or starts with the template marker CHANGEME_. There is no environment override for graph credentials. An explicit environ mapping can be passed for tests. The config subcommand of the cli prints the result.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

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

"""Configuration loader for the Volk Test Harness."""

from __future__ import annotations

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

import yaml

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.harness.models import HarnessConfig

_logger = logging.getLogger(__name__)

DEFAULT_CONFIG_PATH = Path("harness/config.yaml")


@traces(spec="102", story="US-001", reqs=["REQ-102-104", "REQ-102-108"])
def load_config(
    path: Path | None = None, environ: Mapping[str, str] | None = None
) -> HarnessConfig:
    """Load harness configuration from YAML and environment variables.

    Priority (highest to lowest):
    1. Environment variables (OS environ)
    2. YAML config file values
    3. HarnessConfig defaults

    Args:
        path: Path to YAML config. Defaults to ``harness/config.yaml``.
        environ: Optional explicit environment mapping (for testing); when
            omitted, reads the real process environment.

    Returns:
        Populated :class:`HarnessConfig` instance.
    """
    if path is None:
        path = DEFAULT_CONFIG_PATH

    # Start with defaults
    config = HarnessConfig()

    # Overlay YAML if present
    if path.exists():
        _logger.debug("Loading harness config from %s", path)
        raw = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
        config = _apply_yaml(config, raw)
    else:
        _logger.warning("Harness config not found at %s — using defaults", path)

    # Overlay environment variables
    config = _apply_env(config, environ=environ)

    # Ensure directories exist
    config.snapshot_dir.mkdir(parents=True, exist_ok=True)
    config.history_dir.mkdir(parents=True, exist_ok=True)
    config.diagnostics_dir.mkdir(parents=True, exist_ok=True)

    return config


@traces(spec="102", story="US-001", reqs=["REQ-102-104"])
def _apply_yaml(config: HarnessConfig, raw: dict[str, Any]) -> HarnessConfig:
    """Overlay YAML values onto an existing config."""
    path_fields = {"snapshot_dir", "history_dir", "diagnostics_dir"}

    for key, value in raw.items():
        if not hasattr(config, key):
            _logger.warning("Unknown config key: %s", key)
            continue
        if key in path_fields and value is not None:
            value = Path(value)
        setattr(config, key, value)

    return config


@traces(spec="102", story="US-001", reqs=["REQ-102-105", "REQ-102-106"])
def _apply_env(
    config: HarnessConfig, environ: Mapping[str, str] | None = None
) -> HarnessConfig:
    """Overlay environment variable values onto config.

    Maps:
    - ``HARNESS_DEFAULT_LLM`` -> ``default_llm``
    - ``HARNESS_REASONING_LLM`` -> ``reasoning_llm``
    - ``REDIS_HOST`` -> ``redis_host``
    - ``REDIS_PORT`` -> ``redis_port``
    - ``KG_ROUTING_URI`` -> ``graph_uri``
    - ``DEEPSEEK_API_KEY`` -> used by judge evaluator (not stored here)

    No ``graph_username``/``graph_password`` override: the ``NEO4J_USERNAME``/
    ``NEO4J_PASSWORD`` env keys were retired with the Neo4j containers on
    2026-07-27, and every remaining graph endpoint negotiates no auth. Those
    fields are only ever set from YAML (or left at their defaults).

    Args:
        config: Config instance to overlay onto.
        environ: Optional explicit environment mapping (for testing); when
            omitted, reads the real process environment.
    """
    env = os.environ if environ is None else environ
    overrides: dict[str, tuple[str, type[Any]]] = {
        "default_llm": ("HARNESS_DEFAULT_LLM", str),
        "reasoning_llm": ("HARNESS_REASONING_LLM", str),
        "redis_host": ("REDIS_HOST", str),
        "redis_port": ("REDIS_PORT", str),
        # KG_ROUTING_URI, not NEO4J_URI (retired 2026-07-27 with the Neo4j
        # containers). No credential overrides: every remaining graph
        # endpoint negotiates no auth.
        "graph_uri": ("KG_ROUTING_URI", str),
    }

    for attr, (env_var, typ) in overrides.items():
        val = env.get(env_var)
        # An unsubstituted .env template sentinel (CHANGEME_*) means "not
        # configured" — same convention as kernel/cli_main._load_config;
        # without the guard the sentinel would surface as e.g. the LLM name
        # in the harness config printout.
        if val is None or val.startswith("CHANGEME_"):
            continue
        if typ is int:
            try:
                setattr(config, attr, int(val))
            except ValueError:
                _logger.warning("Invalid integer for %s: %s", env_var, val)
        else:
            setattr(config, attr, val)

    return config